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

cassandra-core/build.gradle.kts

@@ -0,0 +1,42 @@
+/*
+ * 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.
+ */
+
+plugins {
+    id("pulsar-connectors.java-conventions")
+}
+dependencies {
+    // Deliberately not a NAR module. nar-conventions disables the jar task and replaces this
+    // project's outgoing artifacts with the NAR itself, so a NAR module cannot be depended on — the
+    // consumer would bundle a .nar inside META-INF/bundled-dependencies where the classloader cannot
+    // reach the classes. The shared sink machinery therefore lives here, in a plain jar that the
+    // `cassandra`, `cassandra-generic-record` and `cassandra-json` NARs each depend on. This mirrors
+    // jdbc/core and its per-database NAR modules.
+    api(libs.pulsar.io.core)
+    api(libs.pulsar.io.common)
+    api(libs.pulsar.client.api)
+    api(libs.cassandra.driver)
+    implementation(libs.jackson.databind)
+    implementation(libs.jackson.dataformat.yaml)
+    implementation(libs.commons.beanutils)
+
+    testImplementation(libs.testcontainers.cassandra)
+    // Test only: builds the Avro-backed GenericRecord that CassandraGenericRecordSink consumes, the
+    // same way the hbase and jdbc sink tests do.
+    testImplementation(libs.pulsar.client)
+}

cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/CassandraGenericRecordSink.java

@@ -0,0 +1,48 @@
+/*
+ * 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 org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.functions.api.Record;
+import org.apache.pulsar.io.cassandra.util.GenericRecordWrapper;
+import org.apache.pulsar.io.cassandra.util.RecordWrapper;
+import org.apache.pulsar.io.core.annotations.Connector;
+import org.apache.pulsar.io.core.annotations.IOType;
+
+/**
+ * Cassandra sink for topics carrying a schema, e.g. Avro or JSON with a registered schema.
+ *
+ * <p>Each field of the incoming {@link GenericRecord} is matched by name against a column of
+ * the target table, so the table definition decides what is written rather than a configured
+ * key/value column pair.
+ */
+@Connector(
+    name = "cassandra-generic-record",
+    type = IOType.SINK,
+    help = "Writes schema-carrying records to Cassandra, mapping record fields onto table columns by name. "
+            + "The target table's columns must all be of type text, varchar, ascii, int, double, float or boolean; "
+            + "any other column type is rejected when the sink starts.",
+    configClass = CassandraSinkConfig.class)
+public class CassandraGenericRecordSink extends CassandraTableSink<GenericRecord> {
+
+    @Override
+    RecordWrapper<GenericRecord> wrapRecord(Record<GenericRecord> record) {
+        return new GenericRecordWrapper(record.getValue());
+    }
+}

cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/CassandraJsonStringSink.java

@@ -0,0 +1,51 @@
+/*
+ * 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 org.apache.pulsar.functions.api.Record;
+import org.apache.pulsar.io.cassandra.util.RecordWrapper;
+import org.apache.pulsar.io.cassandra.util.StringRecordWrapper;
+import org.apache.pulsar.io.core.annotations.Connector;
+import org.apache.pulsar.io.core.annotations.IOType;
+
+/**
+ * Cassandra sink for topics carrying raw JSON strings with no registered schema.
+ *
+ * <p>Each message value is parsed as a JSON object and its top-level fields are matched by
+ * name against the columns of the target table.
+ *
+ * <p>This is distinct from the {@code cassandra} sink ({@link CassandraStringSink}), which
+ * treats the message as an opaque string and writes it to a single configured column. A
+ * deployment wanting JSON-to-column mapping has to select this sink explicitly; the
+ * behaviour of the existing {@code cassandra} sink is unchanged.
+ */
+@Connector(
+    name = "cassandra-json",
+    type = IOType.SINK,
+    help = "Writes raw JSON string messages to Cassandra, mapping top-level JSON fields onto table columns by name. "
+            + "The target table's columns must all be of type text, varchar, ascii, int, double, float or boolean; "
+            + "any other column type is rejected when the sink starts.",
+    configClass = CassandraSinkConfig.class)
+public class CassandraJsonStringSink extends CassandraTableSink<String> {
+
+    @Override
+    RecordWrapper<String> wrapRecord(Record<String> record) {
+        return new StringRecordWrapper(record.getValue());
+    }
+}

cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraSinkConfig.java → cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/CassandraSinkConfig.java

@@ -60,20 +60,27 @@ public class CassandraSinkConfig implements Serializable {
         defaultValue = "",
         help = "The key space used for writing pulsar messages to")
     private String keyspace;
+    // Not required = true: that is enforced at load time for every sink sharing this config, and the
+    // table-mapping sinks have no use for it. `cassandra` still requires it, and checks so itself in
+    // CassandraAbstractSink.open().
     @FieldDoc(
-        required = true,
+        required = false,
         defaultValue = "",
-        help = "The key name of the cassandra column family")
+        help = "The key name of the cassandra column family. Required by the `cassandra` sink. "
+                + "Unused by `cassandra-generic-record` and `cassandra-json`, which map record "
+                + "fields onto columns by name.")
     private String keyname;
     @FieldDoc(
         required = true,
         defaultValue = "",
         help = "The cassandra column family name")
     private String columnFamily;
     @FieldDoc(
-        required = true,
+        required = false,
         defaultValue = "",
-        help = "The column name of the cassandra column family")
+        help = "The column name of the cassandra column family. Required by the `cassandra` sink. "
+                + "Unused by `cassandra-generic-record` and `cassandra-json`, which map record "
+                + "fields onto columns by name.")
     private String columnName;
 
     public static CassandraSinkConfig load(String yamlFile) throws IOException {

cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/CassandraTableSink.java

@@ -0,0 +1,151 @@
+/*
+ * 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 com.datastax.driver.core.BoundStatement;
+import com.datastax.driver.core.PreparedStatement;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.ResultSetFuture;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.MoreExecutors;
+import java.util.Map;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.functions.api.Record;
+import org.apache.pulsar.io.cassandra.util.BoundStatementProvider;
+import org.apache.pulsar.io.cassandra.util.CassandraConnector;
+import org.apache.pulsar.io.cassandra.util.RecordWrapper;
+import org.apache.pulsar.io.cassandra.util.TableMetadataProvider;
+import org.apache.pulsar.io.common.IOConfigUtils;
+import org.apache.pulsar.io.core.Sink;
+import org.apache.pulsar.io.core.SinkContext;
+
+/**
+ * Base class for Cassandra sinks that map a structured record onto the columns of the
+ * target table.
+ *
+ * <p>Unlike {@link CassandraAbstractSink}, which writes a fixed key/value column pair named
+ * by {@code keyname} and {@code columnName}, this base reads the table definition from the
+ * cluster metadata and binds every column it can find a matching field for. Subclasses only
+ * have to say how to read a field out of their record type, by returning the appropriate
+ * {@link RecordWrapper}.
+ *
+ * <p>{@link CassandraAbstractSink} and its {@code cassandra} sink are deliberately left
+ * alone: their key/value behaviour is what existing deployments are configured against.
+ */
+@Slf4j
+public abstract class CassandraTableSink<T> implements Sink<T> {
+
+    CassandraConnector connector;
+    CassandraSinkConfig cassandraSinkConfig;
+    PreparedStatement stmt;
+    BoundStatementProvider boundStatementProvider;
+
+    @Override
+    public void open(Map<String, Object> config, SinkContext ctx) throws Exception {
+
+        cassandraSinkConfig = IOConfigUtils.loadWithSecrets(config, CassandraSinkConfig.class, ctx);
+
+        if (cassandraSinkConfig.getRoots() == null
+                || cassandraSinkConfig.getKeyspace() == null
+                || cassandraSinkConfig.getColumnFamily() == null) {
+            throw new IllegalArgumentException("Required property not set.");
+        }
+        cassandraSinkConfig.validateCredentials();
+
+        connector = new CassandraConnector(cassandraSinkConfig);
+        connector.connect();
+
+        TableMetadataProvider.TableDefinition table = TableMetadataProvider.getTableDefinition(
+                connector.getTableMetadata(),
+                cassandraSinkConfig.getKeyspace(),
+                cassandraSinkConfig.getColumnFamily());
+        rejectUnsupportedColumnTypes(table);
+        boundStatementProvider = new BoundStatementProvider(table);
+    }
+
+    @Override
+    public void write(Record<T> record) throws Exception {
+
+        BoundStatement bs;
+        try {
+            bs = boundStatementProvider.bindStatement(getStatement(), wrapRecord(record));
+        } catch (Exception e) {
+            // Everything before the statement is handed to the driver can fail on the record's own
+            // content: malformed JSON, a null value, a field that will not coerce to its column's type.
+            // Without this the exception leaves write() having neither acked nor failed the record, so
+            // the sink dies, Pulsar redelivers the same message, and one bad message becomes a restart
+            // loop. Failing it explicitly keeps the poison message the broker's problem, not ours.
+            log.error("Discarding record that could not be bound to {}.{}",
+                    cassandraSinkConfig.getKeyspace(), cassandraSinkConfig.getColumnFamily(), e);
+            record.fail();
+            return;
+        }
+
+        ResultSetFuture future = connector.getSession().executeAsync(bs);
+
+        Futures.addCallback(future,
+                new FutureCallback<ResultSet>() {
+                    @Override
+                    public void onSuccess(ResultSet result) {
+                        record.ack();
+                    }
+
+                    @Override
+                    public void onFailure(Throwable t) {
+                        record.fail();
+                    }
+                }, MoreExecutors.directExecutor());
+    }
+
+    @Override
+    public void close() {
+        if (connector != null) {
+            connector.close();
+        }
+    }
+
+    /**
+     * Refuses a table holding a column this sink cannot bind a value to. Every column is a candidate
+     * for every record, so one unsupported column means writes fail — either on every record, or, for
+     * the integer widths, on whichever records happen to carry a value of the wrong magnitude. Saying
+     * so at {@code open()} names the column and its type once, instead of leaving an operator to read
+     * an {@code InvalidTypeException} per message.
+     */
+    private void rejectUnsupportedColumnTypes(TableMetadataProvider.TableDefinition table) {
+        for (TableMetadataProvider.ColumnId column : table.getColumns()) {
+            if (!RecordWrapper.supports(column.getType())) {
+                throw new IllegalArgumentException("Column '" + column.getName() + "' of "
+                        + cassandraSinkConfig.getKeyspace() + "." + cassandraSinkConfig.getColumnFamily()
+                        + " has type '" + column.getType()
+                        + "', which this sink cannot map a record field onto. Supported column types: "
+                        + RecordWrapper.supportedColumnTypes() + ".");
+            }
+        }
+    }
+
+    abstract RecordWrapper<T> wrapRecord(Record<T> record);
+
+    PreparedStatement getStatement() {
+        if (stmt == null) {
+            stmt = connector.getPreparedStatement();
+        }
+        return stmt;
+    }
+}

cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/util/BoundStatementProvider.java

@@ -0,0 +1,61 @@
+/*
+ * 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.util;
+
+import com.datastax.driver.core.BoundStatement;
+import com.datastax.driver.core.PreparedStatement;
+
+@SuppressWarnings("rawtypes")
+public class BoundStatementProvider {
+    final TableMetadataProvider.TableDefinition tableDefinition;
+
+    public BoundStatementProvider(TableMetadataProvider.TableDefinition tableDefinition) {
+        this.tableDefinition = tableDefinition;
+    }
+
+    public BoundStatement bindStatement(PreparedStatement stmt, RecordWrapper wrapper) {
+        Object[] boundValues = new Object[tableDefinition.getColumns().size()];
+        boolean[] supplied = new boolean[boundValues.length];
+        int idx = 0;
+
+        for (TableMetadataProvider.ColumnId column : tableDefinition.getColumns()) {
+            if (wrapper.containsKey(column.getName())) {
+                // A field the record carries as null stays an explicit null: saying a column has no
+                // value is not the same as not mentioning the column.
+                boundValues[idx] = wrapper.get(column);
+                supplied[idx] = true;
+            }
+            idx++;
+        }
+
+        BoundStatement bound = stmt.bind(boundValues);
+        // bind(Object...) writes a real null into every position the record had no field for, and in
+        // Cassandra a null is a deletion: re-writing a row would erase the columns this record says
+        // nothing about, and every insert would leave a tombstone per absent column for compaction to
+        // deal with. Unsetting those positions restores "leave whatever is there alone", which is what
+        // mapping fields onto the columns they match should mean.
+        for (int i = 0; i < supplied.length; i++) {
+            if (!supplied[i]) {
+                bound.unset(i);
+            }
+        }
+        return bound;
+    }
+
+}