[improve][pulsar-io] Added support for generic record and raw JSON string schemas to CassandraSink - #16179
Closed
david-streamlio wants to merge 16 commits into
Closed
[improve][pulsar-io] Added support for generic record and raw JSON string schemas to CassandraSink#16179david-streamlio wants to merge 16 commits into
david-streamlio wants to merge 16 commits into
Conversation
github-actions
Bot
added
the
doc-required
Contributor
Author
|
@tspannhw can you review and upvote? |
|
The pr had no activity for 30 days, mark with Stale label. |
Contributor
Author
|
/pulsarbot run-failure-checks |
Contributor
Author
|
/pulsarbot ready-to-test |
eolivelli
requested changes
Oct 12, 2022
eolivelli
left a comment
eolivelli
left a comment
Contributor
There was a problem hiding this comment.
Great work, thanks
I left some comments, please take a look
Codecov Report
@@ Coverage Diff @@
## master #16179 +/- ##
============================================
- Coverage 46.34% 46.29% -0.05%
- Complexity 10394 10420 +26
============================================
Files 703 703
Lines 68838 68858 +20
Branches 7379 7383 +4
============================================
- Hits 31905 31880 -25
- Misses 33324 33375 +51
+ Partials 3609 3603 -6
Flags with carried forward coverage won't be shown. Click here to find out more.
|
Contributor
Author
|
/pulsarbot run-failure-checks |
Contributor
Author
|
@eolivelli I have made the requested changes, can you PTAL when you get a chance? Thank! |
Contributor
Author
|
@eolivelli , Can you please take a look at this when you get the chance? Thanks again! |
Contributor
Author
|
@eolivelli Can I please get some feedback on these changes I made in response to your initial feedback? Thanks again for the review, I really appreciated it. |
Signed-off-by: tison <wander4096@gmail.com>
tisonkun
previously requested changes
Dec 11, 2022
Member
There was a problem hiding this comment.
@david-streamlio Please fix the compilation failure. It seems that you use JUnit while Pulsar use TestNG as the test platform.
I've pushed a merge commit and fixing for license header to your remote branch so be aware to git pull before working on it.
BTW, please try to expand all star import.
Contributor
Author
|
/pulsarbot run-failure-checks |
Contributor
Author
|
@tisonkun @eolivelli I would appreciate another review when you have the time. |
lhotari
approved these changes
Feb 6, 2026
lhotari
left a comment
lhotari
left a comment
Member
There was a problem hiding this comment.
LGTM. There doesn't seem to be integration tests, but those could be added later.
lhotari
reviewed
Feb 6, 2026
Comment on lines
-35
to
-41
| public class CassandraStringSink extends CassandraAbstractSink<String, String> { | ||
| @Override | ||
| public KeyValue<String, String> extractKeyValue(Record<byte[]> record) { | ||
| String key = record.getKey().orElseGet(() -> new String(record.getValue())); | ||
| return new KeyValue<>(key, new String(record.getValue())); | ||
| RecordWrapper<String> wrapRecord(Record<String> record) { | ||
| return new StringRecordWrapper(record.getValue()); | ||
| } | ||
| } No newline at end of file |
Member
There was a problem hiding this comment.
The behavior of the existing CassandraStringSink shouldn't be modified since it breaks backwards compatibility.
lhotari
requested changes
Feb 6, 2026
lhotari
left a comment
lhotari
left a comment
Member
There was a problem hiding this comment.
The behavior of the existing CassandraStringSink shouldn't be modified since it breaks backwards compatibility. Breaking existing behavior will be surprising for existing users. That's why it's better to add the new behavior only when specific configuration is used.
This was referenced Aug 17, 2026
Contributor
Author
|
Closing this in favour of two PRs against
Why it moved rather than being updated here. PIP-465 removed the IO connectors from this Why two PRs. Re-reading this one, it was doing two independent things under a title that mentions @lhotari — your change request is addressed, and I took it further than asked. You wrote:
You were right, and re-reading the diff the damage was wider than the line you flagged. This PR also Rather than gate the new behaviour behind a configuration flag, #137 makes it a separate connector. One thing in #136 does touch the existing path and is worth your eye: config loading moves from Your other note — that there were no integration tests and they could come later — is partly Thanks for the review. It was the right call and the design is better for it. |
david-streamlio
added a commit
to david-streamlio/pulsar-connectors
that referenced
this pull request
Aug 18, 2026
### 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 is split out separately. ### 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. 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. **Not covered:** no end-to-end test drives `CassandraGenericRecordSink` or `CassandraJsonStringSink` through a running sink — the coverage is of the machinery underneath them. The original PR's answer to this was the IDE harness described above. Happy to add a proper integration test here rather than defer it. ### 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 - [x] 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`, both already in the version catalog. ### Documentation - [ ] `doc-required` - [x] `doc-not-needed` - [ ] `doc` - [ ] `doc-complete` Both new sinks are described by their `@Connector(help = ...)` text.
david-streamlio
added a commit
to david-streamlio/pulsar-connectors
that referenced
this pull request
Aug 18, 2026
### 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 is split out separately. ### 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. 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. **Not covered:** no end-to-end test drives `CassandraGenericRecordSink` or `CassandraJsonStringSink` through a running sink — the coverage is of the machinery underneath them. The original PR's answer to this was the IDE harness described above. Happy to add a proper integration test here rather than defer it. ### 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 - [x] 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`, both already in the version catalog. ### Documentation - [ ] `doc-required` - [x] `doc-not-needed` - [ ] `doc` - [ ] `doc-complete` Both new sinks are described by their `@Connector(help = ...)` text.
david-streamlio
added a commit
to david-streamlio/pulsar-connectors
that referenced
this pull request
Aug 18, 2026
### 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 is split out separately. ### 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. Two defects the end-to-end test below found, both in code this commit 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. 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. `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. Both fixes are mutation-checked: reverting either one fails a test, with `Not a valid schema field` and `NullPointerException` respectively. `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. **Not covered:** no end-to-end test drives `CassandraGenericRecordSink` or `CassandraJsonStringSink` through a running sink — the coverage is of the machinery underneath them. The original PR's answer to this was the IDE harness described above. Happy to add a proper integration test here rather than defer it. ### 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 - [x] 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` - [x] `doc-not-needed` - [ ] `doc` - [ ] `doc-complete` Both new sinks are described by their `@Connector(help = ...)` text.
david-streamlio
added a commit
to apache/pulsar-connectors
that referenced
this pull request
Aug 18, 2026
* [feat][io] Cassandra sink: support authenticated clusters
### Motivation
The Cassandra sink cannot connect to a cluster that requires authentication.
`CassandraSinkConfig` has no credential fields and `CassandraAbstractSink`
builds its `Cluster` with contact points only:
Cluster.Builder b = Cluster.builder();
...
cluster = b.withoutJMXReporting().build();
Any cluster running `PasswordAuthenticator` — which is the normal production
setting — is therefore unusable with this connector, with no workaround short
of forking it.
This gap was first reported in apache/pulsar#16179 (June 2022), which bundled
the fix with a larger schema-mapping feature. That PR could not be rebased
after PIP-465 moved the connectors out of the core repo, and the
authentication half stands on its own, so it is split out here.
### Modifications
- `CassandraSinkConfig` gains `userName` and `password`, both optional and
`sensitive = true`. No existing field is changed.
- `CassandraAbstractSink.createClient` calls `withCredentials(...)` only when
both are supplied, so an unset pair leaves the connection exactly as it was.
- `open()` loads config through `IOConfigUtils.loadWithSecrets` instead of
`CassandraSinkConfig.load`, so the password can be supplied as a Pulsar
secret rather than sitting in plaintext config. This is the same helper the
kinesis, redis, canal and azure-data-explorer connectors already use. Fields
with no secret configured still come from the config map; the pre-existing
`CassandraStringSinkTest`, which opens the sink with a mocked `SinkContext`,
passes unchanged and covers that.
The sink's write path, the `cassandra` connector registration and
`CassandraStringSink` are untouched.
### Verifying this change
`:cassandra:check` passes. `CassandraSinkAuthTest` adds four cases: credentials
survive config loading, they are null when unconfigured, and the sink writes
successfully both with and without them.
Stated plainly, those do not prove the server rejects an unauthenticated
connection: the test container runs the default `AllowAllAuthenticator`, and
switching it to `PasswordAuthenticator` needs a full version-specific
`cassandra.yaml` override. Enforcement is the server's behaviour; what is this
connector's to get right is that credentials reach the driver and that an unset
pair changes nothing. Happy to add a container with authentication enabled if
reviewers would rather have it.
### Does this pull request potentially affect one of the following parts:
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [x] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [x] Anything that affects deployment
**The default values of configurations**: ticked for visibility. No existing
default changes — two new keys are added, both defaulting to unset, where unset
means exactly today's behaviour.
**Deployment**: the module gains `pulsar-io-common`, already in the version
catalog and used by several other connectors. Nothing changes for a deployment
that does not set the new keys.
### Documentation
- [ ] `doc-required`
- [x] `doc-not-needed`
- [ ] `doc`
- [ ] `doc-complete`
The new settings are described by their `@FieldDoc` help text.
* [improve][test] Cassandra sink: assert the server rejects unauthenticated connections
### Motivation
The tests added with the `userName` / `password` settings assert that credentials
survive config loading and that supplying them still yields a working write. They
do not assert that the credentials are needed for anything: the test container
runs the image default `AllowAllAuthenticator`, so a connection with no
credentials is accepted too.
That leaves the settings covered only by their own plumbing. Deleting the
`withCredentials()` call from `CassandraAbstractSink.createClient()` leaves
`CassandraSinkAuthTest` entirely green — all four tests pass with the feature
removed.
The stated reason for stopping there was that enabling `PasswordAuthenticator`
needs a full version-specific `cassandra.yaml`. That is true of the container's
`withConfigurationOverride()`, which volume-maps a whole directory over
`/etc/cassandra` and so requires carrying every file Cassandra reads from there.
Editing the shipped `cassandra.yaml` in place does not.
### Modifications
`CassandraSinkAuthEnforcementTest` runs a container whose `cassandra.yaml` has
been `sed`-ed to `PasswordAuthenticator` before the entrypoint starts, and drives
`CassandraStringSink` against it:
- `serverRejectsUnauthenticatedConnection` connects with the driver alone, no sink
involved, and requires a refusal. Every other assertion in the class is only
worth something while this one holds.
- `sinkFailsToOpenWithoutCredentials` and `sinkFailsToOpenWithWrongPassword`
require `open()` to be refused, unwrapping an `AuthenticationException` reported
either directly or inside a `NoHostAvailableException`.
- `sinkWritesWhenCredentialsAreCorrect` requires the configured pair to get past
that refusal and the row to land.
A `grep -q` after the `sed` makes a substitution that matches nothing fatal: the
container exits rather than starting under `AllowAllAuthenticator` and leaving the
assertions vacuous. Cassandra 5 nests the setting under `authenticator.class_name`,
so a later image tag would otherwise gut this class silently.
`CassandraSinkAuthTest` keeps its unauthenticated container and its tests
unchanged — that an unset credential pair leaves an existing deployment alone
needs a server that does not ask for one. Its scope note now says so and points
here instead of stating the gap.
### Verifying this change
`:cassandra:check` passes, nine tests. With `withCredentials()` disabled in
`CassandraAbstractSink`, `sinkWritesWhenCredentialsAreCorrect` fails while
`CassandraSinkAuthTest` stays green, which is the gap this closes. Breaking the
`sed` pattern on purpose fails container startup rather than any assertion. The
new class ran clean five consecutive times; it costs one additional container.
* [fix][io] Cassandra sink: reject a half-configured credential pair
### Motivation
`createClient()` applies credentials only when both `userName` and `password` are
set, so a pair with one half missing does not fail — it connects unauthenticated.
Against a cluster requiring authentication that surfaces as the driver reporting
that no authenticator was configured, which names neither setting and reads like
the credentials were never supplied at all rather than like the typo it is.
Against a cluster not requiring authentication it succeeds silently, and the
operator learns nothing until the cluster is locked down.
Half a pair is never a meaningful configuration. Whichever half is present, the
intent was to authenticate.
### Modifications
`CassandraSinkConfig.validateCredentials()` rejects the pair with an
`IllegalArgumentException` naming both settings, and `hasCredentials()` answers
whether to authenticate at all. The rule lives on the config rather than in a sink
because it is a property of the settings, not of any one sink, and the sinks added
on top of this one need the same answer.
`CassandraAbstractSink.open()` calls it alongside the existing required-property
validation, and so before any connection is attempted.
The `@FieldDoc` help on both fields now says they are set together.
Tests: `openRejectsUsernameWithoutPassword` and `openRejectsPasswordWithoutUsername`
in `CassandraSinkAuthTest`, and `sinkFailsToOpenWithUnknownUser` in
`CassandraSinkAuthEnforcementTest`, which completes the credential permutations
against an enforcing server — absent, wrong password, unknown user, correct.
### Verifying this change
`:cassandra:check` passes, twelve tests. The two new cases point `roots` at a port
nothing listens on, so they assert the rejection happens before connecting rather
than merely that it happens: with the validation removed they fail with
`NoHostAvailableException: ... Cannot connect` instead of passing for the wrong
reason.
david-streamlio
added a commit
to david-streamlio/pulsar-connectors
that referenced
this pull request
Aug 18, 2026
### 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 is split out separately. ### 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. Two defects the end-to-end test below found, both in code this commit 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. 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. `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. Both fixes are mutation-checked: reverting either one fails a test, with `Not a valid schema field` and `NullPointerException` respectively. `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. **Not covered:** no end-to-end test drives `CassandraGenericRecordSink` or `CassandraJsonStringSink` through a running sink — the coverage is of the machinery underneath them. The original PR's answer to this was the IDE harness described above. Happy to add a proper integration test here rather than defer it. ### 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 - [x] 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` - [x] `doc-not-needed` - [ ] `doc` - [ ] `doc-complete` Both new sinks are described by their `@Connector(help = ...)` text.
david-streamlio
added a commit
to david-streamlio/pulsar-connectors
that referenced
this pull request
Aug 18, 2026
### 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 is split out separately. ### 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. Two defects the end-to-end test below found, both in code this commit 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. 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. `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. Both fixes are mutation-checked: reverting either one fails a test, with `Not a valid schema field` and `NullPointerException` respectively. `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. **Not covered:** no end-to-end test drives `CassandraGenericRecordSink` or `CassandraJsonStringSink` through a running sink — the coverage is of the machinery underneath them. The original PR's answer to this was the IDE harness described above. Happy to add a proper integration test here rather than defer it. ### 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 - [x] 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` - [x] `doc-not-needed` - [ ] `doc` - [ ] `doc-complete` Both new sinks are described by their `@Connector(help = ...)` text.
david-streamlio
added a commit
to david-streamlio/pulsar-connectors
that referenced
this pull request
Aug 18, 2026
### 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 is split out separately. ### 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. The two new sinks therefore get a NAR each, 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 bundles 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 import moves and `sinkConfigClass` keeps its value. Defects found by the end-to-end test below, all in code this commit 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` bound an `Object[]` through `bind(Object...)`, which overwrites the UNSET marker with a real null at every position the record had no field for. In Cassandra a null is a deletion, so re-writing a row erased the columns the new record said nothing about, and every insert left a tombstone per absent column. Those positions are now unset, which is what "map the fields onto the columns they match" should mean. A field carried as null still binds as an explicit null. - `CassandraTableSink.write()` let an exception from `wrapRecord`/`bindStatement` escape, so a record whose content 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, and one poison message became a restart loop. 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. It now guards on the fields. - A bad `keyspace`/`columnFamily` produced a bare `NullPointerException` from `Metadata.getKeyspace(...).getTable(...)`. Both paths now name the setting and value, and `CassandraConnector.getTableFields()` resolves through `TableMetadataProvider` so the INSERT's column list and the binder's positions come from one place. - `open()` now refuses a table holding a column type the wrapper cannot bind — the supported set is text/varchar/ascii, int, double, float and boolean. Types such as `timestamp`, `uuid`, `inet`, `decimal` and `varint` need Java objects neither Jackson nor Avro produces, and `bigint`/`smallint`/`tinyint` fail on value magnitude, since Jackson decodes a small number as an `Integer`. One named error at startup beats an `InvalidTypeException` on every record, or on 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. `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. The fixes are 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 case, and an unacked record for the poison-pill case. `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. **Not covered:** no end-to-end test drives `CassandraGenericRecordSink` or `CassandraJsonStringSink` through a running sink — the coverage is of the machinery underneath them. The original PR's answer to this was the IDE harness described above. Happy to add a proper integration test here rather than defer it. ### 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 - [x] 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` - [x] `doc-not-needed` - [ ] `doc` - [ ] `doc-complete` Both new sinks are described by their `@Connector(help = ...)` text.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
8 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Motivation
The current implementation of the Cassandra Sink connector only supported a single schema type (key, string). This is not useful for production. So I modified the code to be able to support any schema type in Cassandra.
Modifications
Added classes that interrogate the database to determine the schema type at runtime. I also added a framework that will extract the values from the supported incoming schema types (GenericRecord, and String) using the table metadata.
Verifying this change
(Please pick either of the following options)
This change added tests and can be verified as follows:
Added integration tests for testing against a Cassandra database
Does this pull request potentially affect one of the following parts:
If
yeswas chosen, please highlight the changesDocumentation
Check the box below or label this PR directly.
Need to update docs?
doc-required(Your PR needs to update docs and you will update later)