From 5bddb1c1c636bcfc03c378b0d2343ac4ce0fae94 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Mon, 17 Aug 2026 15:13:54 -0700 Subject: [PATCH 1/3] [feat][io] Cassandra sink: support authenticated clusters MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ### 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. --- cassandra/build.gradle.kts | 1 + .../io/cassandra/CassandraAbstractSink.java | 16 +- .../io/cassandra/CassandraSinkConfig.java | 14 ++ .../io/cassandra/CassandraSinkAuthTest.java | 165 ++++++++++++++++++ 4 files changed, 195 insertions(+), 1 deletion(-) create mode 100644 cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthTest.java diff --git a/cassandra/build.gradle.kts b/cassandra/build.gradle.kts index da76cc01b8..5e89fef8ef 100644 --- a/cassandra/build.gradle.kts +++ b/cassandra/build.gradle.kts @@ -23,6 +23,7 @@ plugins { } dependencies { implementation(libs.pulsar.io.core) + implementation(libs.pulsar.io.common) implementation(libs.jackson.databind) implementation(libs.jackson.dataformat.yaml) implementation(libs.cassandra.driver) diff --git a/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraAbstractSink.java b/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraAbstractSink.java index 4f96df280d..9cf28e678e 100644 --- a/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraAbstractSink.java +++ b/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraAbstractSink.java @@ -29,6 +29,7 @@ import com.google.common.util.concurrent.MoreExecutors; import java.util.Map; import org.apache.pulsar.functions.api.Record; +import org.apache.pulsar.io.common.IOConfigUtils; import org.apache.pulsar.io.core.KeyValue; import org.apache.pulsar.io.core.Sink; import org.apache.pulsar.io.core.SinkContext; @@ -47,7 +48,10 @@ public abstract class CassandraAbstractSink implements Sink { @Override public void open(Map config, SinkContext sinkContext) throws Exception { - cassandraSinkConfig = CassandraSinkConfig.load(config); + // loadWithSecrets rather than load(): it resolves @FieldDoc(sensitive = true) fields from the + // connector's secrets provider, so the password need not sit in plaintext config. Fields with + // no secret configured still come from the config map, as before. + cassandraSinkConfig = IOConfigUtils.loadWithSecrets(config, CassandraSinkConfig.class, sinkContext); if (cassandraSinkConfig.getRoots() == null || cassandraSinkConfig.getKeyspace() == null || cassandraSinkConfig.getKeyname() == null @@ -98,10 +102,20 @@ private void createClient(String roots) { b.withPort(Integer.parseInt(hostPort[1])); } } + // Authenticate only when credentials were supplied; an unset pair leaves the connection + // exactly as it was before these settings existed. + if (hasText(cassandraSinkConfig.getUserName()) + && hasText(cassandraSinkConfig.getPassword())) { + b.withCredentials(cassandraSinkConfig.getUserName(), cassandraSinkConfig.getPassword()); + } cluster = b.withoutJMXReporting().build(); session = cluster.connect(); session.execute("USE " + cassandraSinkConfig.getKeyspace()); } public abstract KeyValue extractKeyValue(Record record); + + private static boolean hasText(String value) { + return value != null && !value.trim().isEmpty(); + } } \ No newline at end of file diff --git a/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraSinkConfig.java b/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraSinkConfig.java index 1dfc69b4d1..b0697ee6f7 100644 --- a/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraSinkConfig.java +++ b/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraSinkConfig.java @@ -34,6 +34,20 @@ public class CassandraSinkConfig implements Serializable { private static final long serialVersionUID = 1L; + @FieldDoc( + required = false, + defaultValue = "", + sensitive = true, + help = "Username used to authenticate against the cluster specified by `roots`. " + + "Leave unset for a cluster that does not require authentication.") + private String userName; + @FieldDoc( + required = false, + defaultValue = "", + sensitive = true, + help = "Password used to authenticate against the cluster specified by `roots`. " + + "Leave unset for a cluster that does not require authentication.") + private String password; @FieldDoc( required = true, defaultValue = "", diff --git a/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthTest.java b/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthTest.java new file mode 100644 index 0000000000..3e4e2d7fc6 --- /dev/null +++ b/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthTest.java @@ -0,0 +1,165 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.io.cassandra; + +import static org.mockito.Mockito.mock; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNull; +import com.datastax.driver.core.Cluster; +import com.datastax.driver.core.Session; +import java.time.Duration; +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import org.apache.pulsar.functions.api.Record; +import org.apache.pulsar.io.core.SinkContext; +import org.testcontainers.containers.CassandraContainer; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +/** + * Covers the {@code userName} / {@code password} settings on {@link CassandraSinkConfig}. + * + *

Scope, stated plainly: these assert that credentials are carried through config loading and + * that supplying them still yields a working connection and write. They do not assert that + * the server rejects a connection without them — the Cassandra test container runs with the default + * {@code AllowAllAuthenticator}, and switching it to {@code PasswordAuthenticator} needs a full + * version-specific {@code cassandra.yaml} override. Enforcement is the server's behaviour; what is + * this connector's to get right is that the credentials reach the driver, and that an unset pair + * leaves the connection exactly as it was. + */ +public class CassandraSinkAuthTest { + + private static final String KEYSPACE = "auth_test_ks"; + private static final String TABLE = "auth_test_table"; + private static final String KEY_COLUMN = "key"; + private static final String VALUE_COLUMN = "value"; + + private CassandraContainer cassandraContainer; + + @BeforeClass + public void setUp() { + cassandraContainer = new CassandraContainer<>("cassandra:4.1") + .withStartupTimeout(Duration.ofMinutes(3)); + cassandraContainer.start(); + + try (Cluster cluster = cassandraContainer.getCluster(); + Session session = cluster.connect()) { + session.execute("CREATE KEYSPACE " + KEYSPACE + + " WITH replication = {'class':'SimpleStrategy', 'replication_factor':'1'}"); + session.execute("CREATE TABLE " + KEYSPACE + "." + TABLE + + " (" + KEY_COLUMN + " text PRIMARY KEY, " + VALUE_COLUMN + " text)"); + } + } + + @AfterClass(alwaysRun = true) + public void tearDown() { + if (cassandraContainer != null) { + cassandraContainer.stop(); + cassandraContainer = null; + } + } + + @Test + public void credentialsSurviveConfigLoading() throws Exception { + Map config = baseConfig(); + config.put("userName", "cassandra"); + config.put("password", "cassandra"); + + CassandraSinkConfig loaded = CassandraSinkConfig.load(config); + + assertEquals(loaded.getUserName(), "cassandra"); + assertEquals(loaded.getPassword(), "cassandra"); + } + + @Test + public void credentialsAreUnsetWhenNotConfigured() throws Exception { + CassandraSinkConfig loaded = CassandraSinkConfig.load(baseConfig()); + + assertNull(loaded.getUserName()); + assertNull(loaded.getPassword()); + } + + @Test + public void sinkWritesWithCredentialsSupplied() throws Exception { + Map config = baseConfig(); + config.put("userName", cassandraContainer.getUsername()); + config.put("password", cassandraContainer.getPassword()); + + assertWriteSucceeds(config, "with-credentials"); + } + + @Test + public void sinkWritesWithoutCredentials() throws Exception { + assertWriteSucceeds(baseConfig(), "no-credentials"); + } + + private void assertWriteSucceeds(Map config, String key) throws Exception { + CassandraStringSink sink = new CassandraStringSink(); + try { + sink.open(config, mock(SinkContext.class)); + + CompletableFuture acked = new CompletableFuture<>(); + sink.write(new Record() { + @Override + public Optional getKey() { + return Optional.of(key); + } + + @Override + public byte[] getValue() { + return ("value-" + key).getBytes(); + } + + @Override + public void ack() { + acked.complete(null); + } + + @Override + public void fail() { + acked.completeExceptionally(new RuntimeException("Record failed")); + } + }); + acked.get(); + } finally { + sink.close(); + } + + try (Cluster cluster = cassandraContainer.getCluster(); + Session session = cluster.connect(KEYSPACE)) { + assertEquals(session + .execute("SELECT * FROM " + TABLE + " WHERE " + KEY_COLUMN + " = '" + key + "'") + .one() + .getString(VALUE_COLUMN), "value-" + key); + } + } + + private Map baseConfig() { + Map config = new HashMap<>(); + config.put("roots", cassandraContainer.getHost() + ":" + cassandraContainer.getMappedPort(9042)); + config.put("keyspace", KEYSPACE); + config.put("keyname", KEY_COLUMN); + config.put("columnFamily", TABLE); + config.put("columnName", VALUE_COLUMN); + return config; + } +} From 3efa568476f15d3d1681cd69f0719d2b6aeeba61 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Mon, 17 Aug 2026 18:30:08 -0700 Subject: [PATCH 2/3] [improve][test] Cassandra sink: assert the server rejects unauthenticated connections MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ### 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. --- .../CassandraSinkAuthEnforcementTest.java | 239 ++++++++++++++++++ .../io/cassandra/CassandraSinkAuthTest.java | 20 +- 2 files changed, 251 insertions(+), 8 deletions(-) create mode 100644 cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthEnforcementTest.java diff --git a/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthEnforcementTest.java b/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthEnforcementTest.java new file mode 100644 index 0000000000..ed2bf7b060 --- /dev/null +++ b/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthEnforcementTest.java @@ -0,0 +1,239 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.io.cassandra; + +import static org.mockito.Mockito.mock; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.fail; +import com.datastax.driver.core.Cluster; +import com.datastax.driver.core.Session; +import com.datastax.driver.core.exceptions.AuthenticationException; +import com.datastax.driver.core.exceptions.NoHostAvailableException; +import java.time.Duration; +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import org.apache.pulsar.functions.api.Record; +import org.apache.pulsar.io.core.SinkContext; +import org.testcontainers.containers.CassandraContainer; +import org.testcontainers.containers.wait.strategy.LogMessageWaitStrategy; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.containers.wait.strategy.WaitAllStrategy; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +/** + * Drives {@link CassandraStringSink} against a Cassandra that actually enforces authentication, and + * so asserts what {@link CassandraSinkAuthTest} cannot: that a connection without credentials — or + * with the wrong ones — is refused by the server, and that the credentials the sink is configured + * with are what gets it past that refusal. + * + *

The container runs {@code PasswordAuthenticator} instead of the image default + * {@code AllowAllAuthenticator}. It gets there by editing the shipped {@code cassandra.yaml} in + * place rather than by mounting a replacement: {@code withConfigurationOverride()} maps a whole + * directory over {@code /etc/cassandra}, which would mean carrying a complete, version-matched copy + * of every file Cassandra reads from there, while a one-line {@code sed} touches only the setting + * under test. The {@code grep} that follows it makes a no-op substitution fatal — if a future image + * tag expresses the authenticator differently, the container fails to start rather than quietly + * reverting to {@code AllowAllAuthenticator} and leaving every assertion here vacuous. + * {@link #serverRejectsUnauthenticatedConnection()} is the second half of that guard: it fails the + * suite if the server ever stops enforcing, whatever the reason. + */ +public class CassandraSinkAuthEnforcementTest { + + private static final String IMAGE = "cassandra:4.1"; + private static final String CONF = "/etc/cassandra/cassandra.yaml"; + private static final String SUPERUSER = "cassandra"; + private static final String SUPERUSER_PASSWORD = "cassandra"; + + private static final String ENABLE_PASSWORD_AUTHENTICATION = + "sed -i 's/^authenticator:.*/authenticator: PasswordAuthenticator/' " + CONF + + " && grep -q '^authenticator: PasswordAuthenticator' " + CONF + + " && exec docker-entrypoint.sh cassandra -f"; + + private static final String KEYSPACE = "auth_enforced_ks"; + private static final String TABLE = "auth_enforced_table"; + private static final String KEY_COLUMN = "key"; + private static final String VALUE_COLUMN = "value"; + + private CassandraContainer cassandraContainer; + + @BeforeClass + public void setUp() { + cassandraContainer = new CassandraContainer<>(IMAGE) + .withCommand("sh", "-c", ENABLE_PASSWORD_AUTHENTICATION) + // The default superuser role is created on a delay after the ring is joined, which is + // later than the CQL port opening; waiting for the port alone would let a test connect + // before there is anything to authenticate against. + .waitingFor(new WaitAllStrategy() + .withStrategy(Wait.forListeningPort()) + .withStrategy(new LogMessageWaitStrategy() + .withRegEx(".*Created default superuser role.*\\n")) + .withStartupTimeout(Duration.ofMinutes(3))); + cassandraContainer.start(); + + try (Cluster cluster = authenticatedCluster(); + Session session = cluster.connect()) { + session.execute("CREATE KEYSPACE " + KEYSPACE + + " WITH replication = {'class':'SimpleStrategy', 'replication_factor':'1'}"); + session.execute("CREATE TABLE " + KEYSPACE + "." + TABLE + + " (" + KEY_COLUMN + " text PRIMARY KEY, " + VALUE_COLUMN + " text)"); + } + } + + @AfterClass(alwaysRun = true) + public void tearDown() { + if (cassandraContainer != null) { + cassandraContainer.stop(); + cassandraContainer = null; + } + } + + /** + * Establishes that the server under test enforces authentication at all, using the driver + * directly so that nothing about the sink is involved. Every other assertion in this class is + * only worth something while this one holds. + */ + @Test + public void serverRejectsUnauthenticatedConnection() { + try (Cluster cluster = cassandraContainer.getCluster()) { + cluster.connect(); + fail("Expected the server to reject a connection carrying no credentials"); + } catch (Exception e) { + assertNotNull(authenticationFailure(e), "Not an authentication failure: " + e); + } + } + + @Test + public void sinkFailsToOpenWithoutCredentials() { + assertOpenIsRejected(baseConfig()); + } + + @Test + public void sinkFailsToOpenWithWrongPassword() { + Map config = baseConfig(); + config.put("userName", SUPERUSER); + config.put("password", "not-the-password"); + + assertOpenIsRejected(config); + } + + @Test + public void sinkWritesWhenCredentialsAreCorrect() throws Exception { + Map config = baseConfig(); + config.put("userName", SUPERUSER); + config.put("password", SUPERUSER_PASSWORD); + + String key = "authenticated"; + CassandraStringSink sink = new CassandraStringSink(); + try { + sink.open(config, mock(SinkContext.class)); + + CompletableFuture acked = new CompletableFuture<>(); + sink.write(new Record() { + @Override + public Optional getKey() { + return Optional.of(key); + } + + @Override + public byte[] getValue() { + return ("value-" + key).getBytes(); + } + + @Override + public void ack() { + acked.complete(null); + } + + @Override + public void fail() { + acked.completeExceptionally(new RuntimeException("Record failed")); + } + }); + acked.get(); + } finally { + sink.close(); + } + + try (Cluster cluster = authenticatedCluster(); + Session session = cluster.connect(KEYSPACE)) { + assertEquals(session + .execute("SELECT * FROM " + TABLE + " WHERE " + KEY_COLUMN + " = '" + key + "'") + .one() + .getString(VALUE_COLUMN), "value-" + key); + } + } + + private void assertOpenIsRejected(Map config) { + CassandraStringSink sink = new CassandraStringSink(); + try { + sink.open(config, mock(SinkContext.class)); + fail("Expected open() to be rejected by the server"); + } catch (Exception e) { + assertNotNull(authenticationFailure(e), "Not an authentication failure: " + e); + } + } + + /** + * Finds the {@link AuthenticationException} the driver raised, or {@code null} if the failure was + * something else. The driver reports an unusable contact point either directly or wrapped in a + * {@link NoHostAvailableException} holding the per-host cause, so both shapes are unwrapped here + * rather than asserting on one and hoping. + */ + private static AuthenticationException authenticationFailure(Throwable t) { + for (Throwable current = t; current != null; current = current.getCause()) { + if (current instanceof AuthenticationException) { + return (AuthenticationException) current; + } + if (current instanceof NoHostAvailableException) { + for (Throwable hostError : ((NoHostAvailableException) current).getErrors().values()) { + AuthenticationException found = authenticationFailure(hostError); + if (found != null) { + return found; + } + } + } + } + return null; + } + + private Cluster authenticatedCluster() { + return Cluster.builder() + .addContactPoint(cassandraContainer.getHost()) + .withPort(cassandraContainer.getMappedPort(CassandraContainer.CQL_PORT)) + .withCredentials(SUPERUSER, SUPERUSER_PASSWORD) + .withoutJMXReporting() + .build(); + } + + private Map baseConfig() { + Map config = new HashMap<>(); + config.put("roots", cassandraContainer.getHost() + ":" + + cassandraContainer.getMappedPort(CassandraContainer.CQL_PORT)); + config.put("keyspace", KEYSPACE); + config.put("keyname", KEY_COLUMN); + config.put("columnFamily", TABLE); + config.put("columnName", VALUE_COLUMN); + return config; + } +} diff --git a/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthTest.java b/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthTest.java index 3e4e2d7fc6..95cc955e13 100644 --- a/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthTest.java +++ b/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthTest.java @@ -36,15 +36,19 @@ import org.testng.annotations.Test; /** - * Covers the {@code userName} / {@code password} settings on {@link CassandraSinkConfig}. + * Covers the {@code userName} / {@code password} settings on {@link CassandraSinkConfig} against a + * cluster that does not require authentication — the image default, + * {@code AllowAllAuthenticator}. * - *

Scope, stated plainly: these assert that credentials are carried through config loading and - * that supplying them still yields a working connection and write. They do not assert that - * the server rejects a connection without them — the Cassandra test container runs with the default - * {@code AllowAllAuthenticator}, and switching it to {@code PasswordAuthenticator} needs a full - * version-specific {@code cassandra.yaml} override. Enforcement is the server's behaviour; what is - * this connector's to get right is that the credentials reach the driver, and that an unset pair - * leaves the connection exactly as it was. + *

What that arrangement is good for is the compatibility half of these settings: credentials + * survive config loading, an unset pair is still unset, and neither supplying credentials nor + * omitting them stops the sink writing to a cluster that never asked for them. The last of those is + * the regression this connector most needs guarded — an existing unauthenticated deployment must be + * unaffected by the settings existing. + * + *

That a server actually refuses a connection without credentials, and that the configured pair + * is what gets past the refusal, is asserted in {@link CassandraSinkAuthEnforcementTest}, which runs + * a container with {@code PasswordAuthenticator} enabled. */ public class CassandraSinkAuthTest { From fef47281ae49e2c9706daa1c79953258d9884103 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Mon, 17 Aug 2026 18:42:33 -0700 Subject: [PATCH 3/3] [fix][io] Cassandra sink: reject a half-configured credential pair MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ### 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. --- .../io/cassandra/CassandraAbstractSink.java | 8 +--- .../io/cassandra/CassandraSinkConfig.java | 33 +++++++++++++- .../CassandraSinkAuthEnforcementTest.java | 9 ++++ .../io/cassandra/CassandraSinkAuthTest.java | 45 +++++++++++++++++++ 4 files changed, 87 insertions(+), 8 deletions(-) diff --git a/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraAbstractSink.java b/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraAbstractSink.java index 9cf28e678e..7653107c67 100644 --- a/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraAbstractSink.java +++ b/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraAbstractSink.java @@ -59,6 +59,7 @@ public void open(Map config, SinkContext sinkContext) throws Exc || cassandraSinkConfig.getColumnName() == null) { throw new IllegalArgumentException("Required property not set."); } + cassandraSinkConfig.validateCredentials(); createClient(cassandraSinkConfig.getRoots()); statement = session.prepare("INSERT INTO " + cassandraSinkConfig.getColumnFamily() + " (" + cassandraSinkConfig.getKeyname() + ", " + cassandraSinkConfig.getColumnName() + ") VALUES (?, ?)"); @@ -104,8 +105,7 @@ private void createClient(String roots) { } // Authenticate only when credentials were supplied; an unset pair leaves the connection // exactly as it was before these settings existed. - if (hasText(cassandraSinkConfig.getUserName()) - && hasText(cassandraSinkConfig.getPassword())) { + if (cassandraSinkConfig.hasCredentials()) { b.withCredentials(cassandraSinkConfig.getUserName(), cassandraSinkConfig.getPassword()); } cluster = b.withoutJMXReporting().build(); @@ -114,8 +114,4 @@ && hasText(cassandraSinkConfig.getPassword())) { } public abstract KeyValue extractKeyValue(Record record); - - private static boolean hasText(String value) { - return value != null && !value.trim().isEmpty(); - } } \ No newline at end of file diff --git a/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraSinkConfig.java b/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraSinkConfig.java index b0697ee6f7..b83af7a4c8 100644 --- a/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraSinkConfig.java +++ b/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraSinkConfig.java @@ -39,14 +39,16 @@ public class CassandraSinkConfig implements Serializable { defaultValue = "", sensitive = true, help = "Username used to authenticate against the cluster specified by `roots`. " - + "Leave unset for a cluster that does not require authentication.") + + "Must be set together with the other half of the pair; leave both unset " + + "for a cluster that does not require authentication.") private String userName; @FieldDoc( required = false, defaultValue = "", sensitive = true, help = "Password used to authenticate against the cluster specified by `roots`. " - + "Leave unset for a cluster that does not require authentication.") + + "Must be set together with the other half of the pair; leave both unset " + + "for a cluster that does not require authentication.") private String password; @FieldDoc( required = true, @@ -83,4 +85,31 @@ public static CassandraSinkConfig load(Map map) throws IOExcepti ObjectMapper mapper = new ObjectMapper(); return mapper.readValue(mapper.writeValueAsString(map), CassandraSinkConfig.class); } + + /** + * Rejects a credential pair with only one half set. 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. Sinks call this before connecting, so the failure names what is wrong. + */ + public void validateCredentials() { + if (hasText(userName) != hasText(password)) { + throw new IllegalArgumentException("userName and password must be supplied together: " + + "set both to authenticate, or neither to connect to a cluster that does not " + + "require authentication."); + } + } + + /** + * Whether to authenticate at all. Only ever true for a complete pair, and + * {@link #validateCredentials()} has already rejected an incomplete one. + */ + public boolean hasCredentials() { + return hasText(userName) && hasText(password); + } + + private static boolean hasText(String value) { + return value != null && !value.trim().isEmpty(); + } } \ No newline at end of file diff --git a/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthEnforcementTest.java b/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthEnforcementTest.java index ed2bf7b060..a7bdfba1d9 100644 --- a/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthEnforcementTest.java +++ b/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthEnforcementTest.java @@ -137,6 +137,15 @@ public void sinkFailsToOpenWithWrongPassword() { assertOpenIsRejected(config); } + @Test + public void sinkFailsToOpenWithUnknownUser() { + Map config = baseConfig(); + config.put("userName", "no-such-user"); + config.put("password", SUPERUSER_PASSWORD); + + assertOpenIsRejected(config); + } + @Test public void sinkWritesWhenCredentialsAreCorrect() throws Exception { Map config = baseConfig(); diff --git a/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthTest.java b/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthTest.java index 95cc955e13..ca9b034d20 100644 --- a/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthTest.java +++ b/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthTest.java @@ -21,6 +21,8 @@ import static org.mockito.Mockito.mock; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertNull; +import static org.testng.Assert.assertTrue; +import static org.testng.Assert.fail; import com.datastax.driver.core.Cluster; import com.datastax.driver.core.Session; import java.time.Duration; @@ -46,6 +48,9 @@ * the regression this connector most needs guarded — an existing unauthenticated deployment must be * unaffected by the settings existing. * + *

It also covers the one case the connector rejects on its own account, without asking a server + * anything: a pair with only one half set. + * *

That a server actually refuses a connection without credentials, and that the configured pair * is what gets past the refusal, is asserted in {@link CassandraSinkAuthEnforcementTest}, which runs * a container with {@code PasswordAuthenticator} enabled. @@ -116,6 +121,46 @@ public void sinkWritesWithoutCredentials() throws Exception { assertWriteSucceeds(baseConfig(), "no-credentials"); } + @Test + public void openRejectsUsernameWithoutPassword() { + Map config = unroutableConfig(); + config.put("userName", "cassandra"); + + assertHalfConfiguredPairRejected(config); + } + + @Test + public void openRejectsPasswordWithoutUsername() { + Map config = unroutableConfig(); + config.put("password", "cassandra"); + + assertHalfConfiguredPairRejected(config); + } + + private void assertHalfConfiguredPairRejected(Map config) { + try { + new CassandraStringSink().open(config, mock(SinkContext.class)); + fail("Expected open() to reject a credential pair with only one half set"); + } catch (IllegalArgumentException e) { + assertTrue(e.getMessage().contains("userName and password must be supplied together"), + "Rejected for some other reason: " + e.getMessage()); + } catch (Exception e) { + fail("Expected IllegalArgumentException, got: " + e); + } + } + + /** + * Config whose {@code roots} point at a port nothing listens on. The rejection above has to + * happen before any connection is attempted, and pointing somewhere unreachable is what makes + * the test say so: drop the validation and this fails connecting rather than passing for the + * wrong reason. + */ + private Map unroutableConfig() { + Map config = baseConfig(); + config.put("roots", "127.0.0.1:1"); + return config; + } + private void assertWriteSucceeds(Map config, String key) throws Exception { CassandraStringSink sink = new CassandraStringSink(); try {