[Feature][Connector-V2] Add Cassandra source connectivity dry-run

[goutamadwant]

Purpose of this pull request

Relates to #10681. Add Cassandra source support for the existing Layer 1 --dry-run connect path through SupportSourceDryRunValidation.

Prepare configured SELECT statements and infer projected schemas from prepared result metadata without executing the application query or constructing a source reader. Share the runtime schema conversion, bound driver operations, close resources, and avoid exposing CQL/server causes in connector failure messages.

Depends on #12429 and remains draft until that prerequisite is merged. The comparison against dev currently includes the bootstrap commit; the dry-run-only change is commit 265e6675b76d4fd9f8c52305f8e68294d058406e. No changes are made to DryRunConnectValidator, shared SPI interfaces, CLI level names or failure semantics. Redis support and later dry-run layers are outside this PR.

Does this PR introduce any user-facing change?

Yes. Cassandra source changes from unsupported/skipped connectivity validation to scoped connection and prepared-schema validation. Normal data-job execution is unchanged. English and Chinese connector and CLI capability documentation is updated.

The dry-run accepts CQL beginning with SELECT, optionally preceded by whitespace. It rejects leading comments, non-SELECT statements, unbound parameters, empty result metadata and duplicate output identifiers. These restrictions apply to dry-run only.

The driver may read system metadata and populate its prepared-statement cache. Success does not prove SELECT permissions, later query execution, consistency-level availability or sink compatibility. Timeouts are per operation, not a single total deadline. No new configuration option, default, dependency or public SPI is introduced.

How was this patch tested?

• Java 8 and Java 11: 32 connector unit tests and 8 live integration cases pass on the stacked branch, including the bootstrap prerequisite's tests.
• Cassandra 4.1.1 coverage includes projected-schema parity, aliases/functions, empty results, multiple tables, missing tables/keyspaces, authenticated success and wrong-password failure.
• Unit tests assert preparation without application-query execution. A live request observer first verifies an ordinary SELECT is observable, then sees no successful executed-CQL callbacks during dry-run; PREPARE is not exposed by that observer, so this is not a wire trace or a claim of zero system-metadata access.
• Packaged CLI success/failure tests pass. The CLI engine uses its Java 8 container; direct factory integration runs in the selected Java 8/11 harness.
• Owned local no-response endpoint probes fail safely on both JDKs. Pre-interrupted calls, cleanup, bounded driver settings and sanitized failures are covered.
• Root formatting and Java 11 ./mvnw -q -DskipTests verify pass. Normal source/sink E2E evidence is recorded separately for the bootstrap prerequisite.

Focused command, run with Java 8 and Java 11 selected through JAVA_HOME:

./mvnw -B -pl seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cassandra-e2e -am \
  -Dtest=CassandraClientTest,CassandraFactoryTest,TypeConvertUtilTest \
  -Dsurefire.failIfNoSpecifiedTests=false -DskipIT=false \
  '-Dit.test=CassandraIT#testDryRun*+testUnavailableBootstrapContactPoint' \
  -DfailIfNoTests=false verify

Check list

• No new binary dependency; license/notice additions are not required.
• English and Chinese documentation updated.
• No existing runtime configuration/API incompatibility or migration required; new dry-run restrictions are documented.
• Existing Cassandra E2E extended. No new connector registration, plugin mapping, distribution entry or CI scope registration is needed.

[DanielLeens]

What Problem Does This PR Solve?

• User pain point: today there is no way to check whether a Cassandra source connection, keyspace/table access, and CQL projection are valid before submitting a real job with --dry-run connect. Users only discover a bad host, wrong keyspace, or a CQL typo when the job actually starts.
• Fix approach: implement SupportSourceDryRunValidation for CassandraSourceFactory. A new CassandraSourceDryRunValidator opens a short-lived, tightly-timeboxed CqlSession, PREPAREs each configured SELECT (never executing it), and reuses the exact same column-to-schema conversion (CassandraSource.buildTableConfig) that a real job uses, so the inferred schema is guaranteed to match runtime. Along the way, CassandraClient.getCqlSessionBuilder is fixed so that all configured hosts become bootstrap contact points on one builder instead of a single randomly chosen host.
• One-sentence summary: adds a metadata-only, credential-safe connectivity/schema preflight for the Cassandra source, reusing the real runtime schema-conversion code path so the dry-run result is trustworthy.

1. Code Change Review

1.1 Core Logic Analysis

CassandraSourceDryRunValidator.inferSchema (new file, source/CassandraSourceDryRunValidator.java:38-95)

• Validates options against CassandraSourceFactory().optionRule(), collects the query list (single cql or each tables_configs[].cql), rejects anything that does not match ^\s*SELECT\s+.+ case-insensitively.
• Opens one CqlSession with a bounded DriverConfigLoader (10s per network wait, RECONNECT_ON_INIT=false, PREPARE_ON_ALL_NODES=false), PREPAREs each query, rejects prepared statements that have unbound variables, and rejects duplicate resulting tableIds.
• Delegates schema construction to CassandraSource.buildTableConfig(String, ColumnDefinitions, String), which was extracted from the existing runtime buildTableConfig overload (source/CassandraSource.java:113-158) specifically so both paths share one conversion routine.

Before/after on the shared helper:

// before: private CassandraTableConfig buildTableConfig(String cql, CqlSession session, ...)
//         executed the CQL, then inlined the ColumnDefinitions -> CatalogTable conversion
// after:  the conversion logic is factored into a package-private static overload
static CassandraTableConfig buildTableConfig(String cql, ColumnDefinitions columnDefs, String keyspace) { ... }

This is a clean, low-risk extraction — same code, now callable from PreparedStatement#getResultSetDefinitions() as well as from an executed ResultSet.

CassandraClient.getCqlSessionBuilder (client/CassandraClient.java:34-52) — real behavior change, not just a refactor:

// before
List<CqlSessionBuilder> list = Arrays.stream(nodeAddress.split(","))
    .map(address -> CqlSession.builder().addContactPoint(<one address>)...)
    .collect(...);
return list.get(ThreadLocalRandom.current().nextInt(list.size()));   // <-- picks ONE address, drops the rest

// after
CqlSessionBuilder builder = CqlSession.builder().withKeyspace(keyspace).withLocalDatacenter(dataCenter);
for (String address : nodeAddress.split(",")) {
    builder.addContactPoint(<address>);                              // <-- ALL addresses on one builder
}
if (!StringUtils.isEmpty(username) || !StringUtils.isEmpty(password)) {
    builder.withAuthCredentials(username, password);
}
return builder;

The pre-existing code was a real bug: with host = "a:9042,b:9042" it built one CqlSessionBuilder per address, each with a single contact point, then threw away all but a randomly-chosen one — so a multi-host host config never actually gave the driver more than one bootstrap contact point, and if that one host happened to be down at connection time, session creation failed outright even though other configured hosts were reachable. The fix registers every configured address as a contact point on a single builder, which is what the Java driver's contact-point mechanism is for. This method is shared by CassandraSource, CassandraSourceReader, CassandraSink, CassandraSinkWriter, and the new dry-run validator, so the fix improves multi-host bootstrap resilience for the existing source and sink as well, not just the new dry-run path.

Key findings:

1. Metadata-only guarantee is real: session.prepare(...) is used, never session.execute(...), and a dedicated IT test (testDryRunPreparesWithoutExecutingApplicationQueries) verifies via a custom RequestTracker that zero requests are tracked for the dry-run path while a control execute() call does register a request — this is a strong, concrete proof rather than a documentation-only claim.
2. Reuse of CassandraSource.buildTableConfig for both the real path and the dry-run path is the right design: it guarantees the dry-run-inferred schema and the runtime schema cannot silently drift apart, and testDryRunSchemaParity (E2E) explicitly asserts the two getProducedCatalogTables() results are equal across several representative queries.
3. The exception handling in inferSchema's try-with-resources catch (RuntimeException failure) (source/CassandraSourceDryRunValidator.java:89-94) discards the original exception's message and cause entirely and replaces it with one static string for every failure mode inside the try block — driver connection failures, IllegalArgumentException("Unbound parameters are not supported"), IllegalArgumentException("Duplicate table identifiers"), CassandraConnectorException("No columns returned by CQL: ..."), and any TypeConvertUtil unsupported-type exception all collapse into the identical message: "Cassandra connect dry-run failed. Check connection, credentials, keyspace, SELECT queries, supported types and unique table identifiers." None of the self-thrown validation messages ("unbound parameters", "duplicate table identifiers") contain literal values or credentials, so there was no need to discard those specific, actionable messages along with genuine driver-exception text. See 1.4 Issue 1 for detail.
4. getCqlSessionBuilder's multi-contact-point fix is bundled into a PR whose title only advertises a dry-run feature. It is well tested (unit test in CassandraClientTest, and IT test testUnavailableBootstrapContactPoint, plus cassandra_to_cassandra.conf was updated to include a deliberately-unreachable cassandra:1 host precisely to exercise the new failover-at-bootstrap behavior), and it is a genuine, in-scope-adjacent bug fix rather than an accidental regression, but it changes production connection behavior for the existing source and sink for every user with a multi-host host config, and that is worth calling out explicitly in the PR description/changelog since it is a behavior change beyond "add dry-run".
5. The DriverConfigLoader used for validation is created and closed independently of the CqlSession (configLoader() at source/CassandraSourceDryRunValidator.java:110-121), and both are acquired in a single try-with-resources with the loader declared first and the session second, so on close the session closes before the loader — correct ordering, since a driver DriverConfigLoader supplied externally via withConfigLoader is not owned by the session and must outlive it. dryRunClosesAndSanitizesPrepareAndCloseFailures additionally verifies that even a session.close() failure occurring during implicit resource cleanup does not leak its (potentially sensitive) message via getSuppressed().

No engine/checkpoint/serialization/state-machine paths are touched — this PR is confined to the Cassandra connector module and the CLI dry-run SPI it opts into, so no mermaid runtime-path diagram is required per the reporting guidance; the call chain is a straightforward DryRunConnectValidator.validateSource → CassandraSourceFactory.inferSchemaForDryRun → CassandraSourceDryRunValidator.inferSchema → CassandraClient.getCqlSessionBuilder/CassandraSource.buildTableConfig.

1.2 Compatibility Impact

Fully compatible. SupportSourceDryRunValidation is a new opt-in interface; connectors that don't implement it keep their existing SKIPPED dry-run status. CassandraSourceFactory's existing optionRule()/createSource() contract, config option names, and defaults are all unchanged. The getCqlSessionBuilder behavior change (all hosts now become contact points instead of one random host) is a backward-compatible reliability improvement: Cassandra contact points are only used for initial cluster discovery, so once connected the driver already learns the full ring topology from whichever contact point it reaches — this change only helps when the previously-chosen single random contact point was unreachable at startup; it does not change consistency levels, replica selection, or steady-state read/write behavior. Still, it is a real behavior change worth a one-line mention in the PR description for changelog purposes since it affects the existing source and sink, not only the new dry-run feature.

1.3 Performance / Side-Effect Analysis

• Bounded blast radius: dedicated DriverConfigLoader caps connection/init-query/set-keyspace/control-connection/metadata-schema-request/request timeouts at 10s each and disables RECONNECT_ON_INIT and PREPARE_ON_ALL_NODES, so a dry-run invocation cannot hang indefinitely or eagerly prepare on every cluster node.
• checkInterrupted() is called before the validation starts and before each query's PREPARE, so a CLI-level interrupt/cancel is honored promptly instead of leaving the process blocked on a slow cluster.
• The session and config loader are always closed via try-with-resources, including on prepare failure — verified by Mockito.verify(fixture.session).close() in every failure-path unit test.
• No new dependency introduced; reuses the existing Cassandra driver already on the classpath for this connector.

1.4 Error Handling and Logging

Issue 1

• Location: seatunnel-connectors-v2/connector-cassandra/src/main/java/org/apache/seatunnel/connectors/seatunnel/cassandra/source/CassandraSourceDryRunValidator.java:89-94
• Problem: The blanket catch (RuntimeException failure) around the whole try-with-resources block replaces every failure — including the two self-thrown, credential-free IllegalArgumentExceptions ("Unbound parameters are not supported", "Duplicate table identifiers"), the pre-existing CassandraConnectorException("No columns returned by CQL: ...") from the shared buildTableConfig, and any TypeConvertUtil unsupported-column-type exception — with one identical generic string. dryRunRejectsMissingColumnsAndUnboundParameters confirms both failure causes produce byte-for-byte the same exception message.
• Risk: Not a correctness bug (pass/fail is still reported correctly, and the centralized DryRunConnectFailureMessageSanitizer used by DryRunConnectValidator.wrap() would already have redacted anything sensitive from a propagated message), but it materially reduces the diagnostic value of a tool whose entire purpose is to let users self-diagnose a bad config before running a real job. A user who mistypes a column, forgets a WHERE binding, or has two tables_configs entries resolving to the same table gets the exact same "check five different things" message as a user with a wrong host or password.
• Best improvement: only discard/replace the message for exceptions that can plausibly carry connection details or literal query values (e.g., failures thrown by the CQL driver itself during session.build()/session.prepare()), and let the two locally-thrown IllegalArgumentExceptions (and other purely structural validation failures raised by this class's own code) propagate with their real, already-safe message — mirroring how KafkaSourceDryRunValidator/S3SourceDryRunValidator let specific exception types propagate and rely on the shared DryRunConnectFailureMessageSanitizer (already used centrally in DryRunConnectValidator.wrap()) for redaction, instead of every connector re-implementing its own blanket message discard.
• Severity: Medium
• Raised by another reviewer: No

2. Code Quality Assessment

2.1 Coding Standards

Every new non-trivial method has a purposeful comment (class-level Javadoc on CassandraSourceDryRunValidator and KafkaSourceDryRunValidator-style single-line intent comments on configLoader(), checkInterrupted(), and the catch block explaining why the message is discarded). The extracted CassandraSource.buildTableConfig(String, ColumnDefinitions, String) overload has no separate Javadoc, but its logic and error paths are unchanged from the original inline code and it is package-private, so this is acceptable. No missing-doc items to flag.

2.2 Test Coverage and Test Stability

Coverage is thorough and exercises the real driver end-to-end, not just mocks:

• Unit tests (CassandraFactoryTest, CassandraClientTest) cover: multi-table dry-run, duplicate-table rejection, non-SELECT/leading-comment rejection before any connection attempt, missing-column/unbound-parameter rejection, pre-existing-interruption handling, message sanitization (including the close()-failure/suppressed-exception edge case), and the multi-contact-point builder behavior (including a duplicate-address case).
• IT tests (CassandraIT) run against a real Testcontainers Cassandra: dry-run vs. real-job schema parity across five representative CQL shapes, zero-application-request proof via a custom RequestTracker, multi-table dry-run against real tables, missing-keyspace and missing-table failure, a full authenticated-connection round trip (including a deliberately wrong password, asserting the failure message does not leak it), a genuinely unreachable bootstrap host used together with a reachable one, and a packaged-CLI --dry-run connect invocation asserting both the success and failure exit codes/output.
• Stability: testDryRunAuthenticatedConnection wraps its first assertion in Awaitility.await().ignoreException(IllegalStateException.class)...untilAsserted(...) to absorb the documented async superuser-creation delay in the Cassandra container — this is a legitimate, bounded (90s) retry for a real environmental race, not a disguised flaky-test workaround, and it is scoped to exactly the exception type expected during that startup window.
• Stability rating: Stable. No sleep-based waits, no unbounded retries, no widened tolerances on unrelated assertions; the one Awaitility retry targets a documented, real startup race and is exception-type-scoped.

2.3 Documentation Updates

docs/en and docs/zh are both updated in lockstep: a new "Connectivity dry-run" section (with a matching translated heading in the zh doc) on the source doc precisely describes the SELECT-only requirement, the "never executes the data query" guarantee, and the caveat that a successful prepare does not prove SELECT permissions or later-execution success; a new note on the host option (source and sink docs, en/zh) accurately documents the multi-contact-point behavior change and is appropriately conservative ("does not change consistency levels or guarantee availability when no suitable node is reachable"); and docs/en(zh)/engines/zeta/user-command.md's dry-run support matrix gains a Cassandra row consistent with the Kafka/S3File rows already present. Documentation matches the code precisely.

3. Architectural Soundness

3.1 Elegance of the Solution

Precise fix. Reusing the existing runtime schema-conversion method (buildTableConfig) rather than writing a parallel schema-inference code path is exactly the right design: it structurally prevents dry-run/runtime schema drift, which is the biggest risk in any dry-run feature. The interface implementation is minimal and follows the same shape already established by KafkaSourceDryRunValidator and S3SourceDryRunValidator.

3.2 Maintainability

Good. The validator is a small, final, stateless utility class with clear single-purpose private helpers. The one maintainability wrinkle is the message-discarding catch block (Issue 1) — a future contributor adding a new validation rule inside the try block will silently lose their specific error message unless they remember to route it around the blanket catch.

3.3 Extensibility

Adding a new rejection rule (e.g., a future column-count cap) fits naturally into the existing per-query loop. The tableIds dedup pattern mirrors the equivalent runtime-path check in CassandraSource.buildTableConfigs, keeping the two paths structurally consistent for future maintenance.

3.4 Historical-Version Compatibility

No config options, defaults, serialization formats, or checkpoint/state contracts are touched. This connector is a bounded-batch source with no checkpoint/restore state, so there is no version-upgrade or savepoint-restore risk to evaluate here.

4. Issue Summary

Number	Issue	Location	Severity
1	Blanket catch (RuntimeException) discards even safe, self-thrown validation messages ("unbound parameters", "duplicate table identifiers"), collapsing all dry-run failure causes into one identical generic message	source/CassandraSourceDryRunValidator.java:89-94	Medium

5. Merge Recommendation

Conclusion: Ready to merge

1. Blockers - must be fixed:
   • None. There is no functional-correctness, compatibility, or data-safety blocker in this diff.
2. Recommended fixes - non-blocking:
   • Issue 1 (Medium): narrow the blanket exception-message discard so the tool's own safe validation messages ("unbound parameters", "duplicate table identifiers") remain visible to the user, consistent with how the Kafka/S3 dry-run validators rely on the shared DryRunConnectFailureMessageSanitizer for redaction instead of a local blanket catch.
   • Please add one line to the PR description calling out that CassandraClient.getCqlSessionBuilder now registers all configured hosts as bootstrap contact points instead of a single randomly-chosen one — this is a genuine, well-tested, backward-compatible reliability fix that benefits the existing source and sink, but reviewers and changelog readers should not have to discover a shared-client behavior change by reading the diff of a PR titled as a dry-run feature.

Overall assessment: this is an excellent, first-review-ready contribution — the dry-run implementation correctly reuses the runtime schema path to guarantee parity, is properly bounded (timeouts, no unbounded reconnect, prompt interruption handling), is proven metadata-only via a request-tracking IT test rather than by assertion alone, and ships thorough bilingual documentation plus both unit and real-container IT coverage, including a legitimate authentication-failure and unreachable-host scenario. The only substantive finding (Issue 1) is a diagnosability trade-off, not a correctness or safety problem, and is easy to address. Note for the maintainer queue: this PR is currently marked as a GitHub draft; it will need to be marked ready for review before it can be merged, and CI on the contributor's fork was still queued at review time (no results yet to evaluate) — please re-check before merging. No alternative architectural approach is warranted here; the chosen "reuse the runtime schema converter, add a bounded metadata-only session" design is the same successful pattern already used by the Kafka and S3File dry-run validators.