[feat][io] Cassandra sink: support authenticated clusters

[david-streamlio]

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() rejects a pair with only one half set, alongside the existing
  required-property validation and so before any connection is attempted. Half a
  pair is never a meaningful configuration — whichever half is present, the intent
  was to authenticate — and connecting unauthenticated instead turns a typo into
  either a rejection from the server that names neither setting, or a silent
  success against a cluster that does not yet require authentication.
• 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, twelve tests across two classes.

CassandraSinkAuthTest runs against an unauthenticated cluster — the image
default, AllowAllAuthenticator — and covers the compatibility half: credentials
survive config loading, they are null when unconfigured, and the sink writes
successfully both with and without them. That last case is the regression this
connector most needs guarded, since an existing unauthenticated deployment must be
unaffected by the settings existing, and it needs a server that does not ask for
credentials in order to mean anything.

CassandraSinkAuthEnforcementTest covers the other half against a cluster running
PasswordAuthenticator:

• 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, sinkFailsToOpenWithWrongPassword and
  sinkFailsToOpenWithUnknownUser 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.

An earlier revision of this description said enforcement could not be asserted
without a full version-specific cassandra.yaml override. That is true only 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 before the entrypoint runs does not,
and is what this does:

.withCommand("sh", "-c",
    "sed -i 's/^authenticator:.*/authenticator: PasswordAuthenticator/' " + CONF
  + " && grep -q '^authenticator: PasswordAuthenticator' " + CONF
  + " && exec docker-entrypoint.sh cassandra -f")

The grep 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 the class silently.

openRejectsUsernameWithoutPassword and openRejectsPasswordWithoutUsername 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.

Three mutation checks, since a passing test proves nothing on its own. With the
withCredentials() call deleted from CassandraAbstractSink.createClient(),
sinkWritesWhenCredentialsAreCorrect fails while CassandraSinkAuthTest stays
entirely green — that gap is precisely what the new class 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.

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

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
• doc-not-needed
• doc
• doc-complete

The new settings are described by their @FieldDoc 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.