[feat][io] Cassandra sink: add generic-record and JSON-string sinks

[david-streamlio]

Motivation

The Cassandra sink can only write one column. CassandraStringSink treats the
message value as an opaque string and writes it to the single column named by
columnName, keyed by keyname. A topic carrying an Avro or JSON schema has
named fields and a table with matching columns, and there is no way to land them
in more than one of those columns. The same applies to raw JSON on a topic with
no registered schema.

This ports the schema-mapping half of apache/pulsar#16179, which predates both
the Gradle migration and PIP-465's move of the connectors to this repository, so
it could not be rebased there. The authentication half went in separately as #136,
and this is now rebased onto it — the diff is schema-mapping-only, as promised
there.

Modifications

Two new sinks, both mapping record fields onto table columns by name, with the
target table's own definition deciding what gets written:

Connector	Input	Behaviour
cassandra (unchanged)	Record<byte[]>	opaque string to the one configured column
cassandra-generic-record (new)	GenericRecord	record fields to columns, matched by name
cassandra-json (new)	String	top-level JSON fields to columns, matched by name

The existing sink is untouched. CassandraStringSink, CassandraAbstractSink
and the cassandra entry in pulsar-io.yaml are unchanged by this commit, so no
existing deployment can be affected. That is a direct response to @lhotari's
review on #16179, which asked that existing behaviour not change and that the new
behaviour be opt-in. The original delivered this by rewriting CassandraStringSink
in place — dropping its @Connector registration, changing the input from
Record<byte[]> to Record<String>, removing the key-falls-back-to-value
behaviour, and reinterpreting the message as JSON. Here the JSON behaviour is its
own connector instead, so there is no configuration path by which the cassandra
sink can start behaving differently.

New machinery, used only by the new sinks:

• CassandraTableSink<T> reads the table definition from cluster metadata at
  open() and binds each column it can find a matching field for.
• util/CassandraConnector owns cluster/session lifecycle and the prepared
  INSERT built from the table's actual column list.
• util/TableMetadataProvider resolves a keyspace/table into its columns,
  partition key and primary key.
• util/BoundStatementProvider binds a wrapped record against that column list.
• util/RecordWrapper and its GenericRecordWrapper / StringRecordWrapper
  implementations read a named field out of each record type and coerce it to the
  column's declared type.

keyname and columnName become required = false on CassandraSinkConfig.
IOConfigUtils.loadWithSecrets enforces required = true for every sink sharing
the config, so leaving them required makes a table-sink config that omits them fail
to load — rejected for missing two settings that sink has no use for, before it can
say so. They are still required by the cassandra sink, which checks both itself in
CassandraAbstractSink.open() and is unchanged; their help text now says which sink
needs them.

CassandraTableSink.open() calls cassandraSinkConfig.validateCredentials() and
CassandraConnector asks hasCredentials(), so the new sinks reject a
half-configured credential pair on the same terms as the existing one rather than
quietly connecting unauthenticated.

Packaging. A NAR's pulsar-io.yaml declares one sinkClass, so one NAR offers one
connector by name. Each new sink therefore gets its own NAR module, and the classes they
share with cassandra move to a plain-jar :cassandra-core that all three NARs depend
on. They cannot depend on :cassandra directly: nar-conventions disables the jar task
and replaces the project's outgoing artifacts with the NAR itself, so a consumer ends up
bundling a .nar inside META-INF/bundled-dependencies, where the classloader cannot
reach the classes. This is the shape jdbc/core and the per-database JDBC NARs already
use. The Java package is unchanged throughout, so no imports move and sinkConfigClass
keeps its value.

Module	Kind	Connector
cassandra-core	plain jar	— shared config, table sink, wrappers, metadata
cassandra	NAR	cassandra (unchanged)
cassandra-generic-record	NAR	cassandra-generic-record
cassandra-json	NAR	cassandra-json

Without this, both new sinks were unreachable by --sink-type — the classes shipped in
the Cassandra NAR but nothing declared them, so only --archive + --classname would
have worked. An earlier revision of this description claimed they simply "ship in the
existing Cassandra NAR", which was true of the class files and misleading about whether
anyone could select them.

Defects found by the end-to-end test, all in code this PR adds:

• GenericRecordWrapper.containsKey() asked getField(name) and treated null as
  absent. GenericAvroRecord.getField() delegates to Avro, which throws
  AvroRuntimeException: Not a valid schema field for a name the schema does not
  carry rather than returning null, so cassandra-generic-record failed on any table
  holding a column the record has no field for — the ordinary case, and the opposite
  of what the sink promises. It now asks the record's schema, via getFields().
• RecordWrapper.getValueAsExpectedType() had no null case, so a field present but
  null threw: NullPointerException on the TEXT branch's toString(), and a
  ConversionException on the numeric ones. A present-but-null field now binds as null.
• BoundStatementProvider was erasing data. It bound an Object[] through
  bind(Object...), which overwrites the UNSET marker with a real null at every position
  the record had no field for — and in Cassandra a null is a deletion. Re-writing a row
  wiped the columns the new record said nothing about, and every insert left a tombstone
  per absent column for compaction. Those positions are now unset. A field carried as
  null still binds as an explicit null, which is a different statement.
• A single bad message could restart-loop the connector. CassandraTableSink.write()
  let an exception from wrapRecord/bindStatement escape, so a record that could not be
  bound — malformed JSON, a null value, a value that will not coerce — was neither acked
  nor failed. The sink died, Pulsar redelivered the same message, repeat. Such a record is
  now failed and logged.
• CassandraConnector.close() called getSession()/getCluster(), which create what
  they return: closing after a failed open() re-attempted the connection and threw a
  second exception, hiding the original cause. It now guards on the fields.
• A bad keyspace/columnFamily produced a bare NullPointerException. Both paths now
  name the setting and its value, and getTableFields() resolves through
  TableMetadataProvider so the INSERT's column list and the binder's positions come from
  one place rather than two independent metadata reads.
• open() now refuses a table holding a column type the wrapper cannot bind. Supported:
  text/varchar/ascii, int, double, float, boolean. timestamp, uuid, inet, decimal
  and varint need Java objects neither Jackson nor Avro produces, and
  bigint/smallint/tinyint fail on magnitude, since Jackson decodes a small number as
  an Integer. One named error at startup beats an InvalidTypeException on every record
  — or, worse, on only some of them.

Deviations from the original, all deliberate:

• close() no longer swallows every Throwable in an empty catch block.
• StringRecordWrapper uses a TypeReference instead of a raw Map.class,
  removing an unchecked-conversion warning.
• AbstractCassandraTest applies init.cql through the Datastax driver rather
  than the container's withInitScript. The script delegate in the
  org.testcontainers.cassandra module is compiled against shaded classes the
  resolved Testcontainers core no longer ships, and fails with
  NoClassDefFoundError; the pre-existing test avoids this by using the older
  org.testcontainers.containers.CassandraContainer, and this now matches it.
• CassandraSinkExec and producers/** are not ported. CassandraSinkExec is an
  IDE-only main() that expects a broker on localhost and then sleeps for ten
  minutes; it would add pulsar-functions-local-runner-original to the test
  classpath for something CI never runs, and the producers exist only to feed it.

Verifying this change

:cassandra:check passes: CassandraConnectorTest asserts the generated INSERT
matches a real table's column list for both a two-column and a seventeen-column
table, TableMetadataProviderTest asserts the resolved table definition, and the
pre-existing CassandraStringSinkTest still passes unchanged.

CassandraSinkConfigValidationTest covers what each sink accepts and rejects at
open() before reaching a cluster: that the new sinks load without keyname and
columnName, that the cassandra sink still refuses to open without either, and
that all of them reject a half-configured credential pair. It points roots at a
port nothing listens on, so a config that gets as far as failing to connect is one
that passed validation — which is what makes the accepting cases mean anything.

CassandraTableSinkIntegrationTest drives both new sinks end to end — real sink,
real cluster, real rows — against airquality.reading from init.cql: seventeen
columns, a compound primary key, and text / int / double / float among them,
of which the records populate seven. It asserts what landed, that a column no field
matched is left null, and that a field no column matched is dropped. The
generic-record case builds its input the way the hbase and jdbc sink tests do —
encode a POJO with AvroSchema, decode with GenericAvroSchema — so the sink sees a
real Avro-backed record rather than a hand-written double, which is what surfaced the
containsKey() defect above.

Every fix is mutation-checked: reverting any one of them fails a test — Not a valid schema field, NullPointerException, expected [CA] but found [null] for the erased
column, an unacked record for the poison-pill case, and with required = false reverted
neither sink opens at all. That all of these were sitting behind the coverage gap this PR
previously deferred is the argument for having closed it rather than shipping on the
machinery tests.

Does this pull request potentially affect one of the following parts:

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Deployment: two new connectors, cassandra-generic-record and
cassandra-json, ship in the existing Cassandra NAR. Nothing changes for a
deployment using the cassandra sink. The module gains pulsar-client-api and
commons-beanutils at compile scope and pulsar-client-original at test scope, all
three already in the version catalog; the test-scope one is not on the NAR's
classpath.

Documentation

• doc-required
• doc-not-needed
• doc
• doc-complete

Both new sinks are described by their @Connector(help = ...) text.

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.