diff --git a/cassandra-core/build.gradle.kts b/cassandra-core/build.gradle.kts new file mode 100644 index 0000000000..202f86b3d4 --- /dev/null +++ b/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) +} diff --git a/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/CassandraGenericRecordSink.java b/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/CassandraGenericRecordSink.java new file mode 100644 index 0000000000..9cde40dc40 --- /dev/null +++ b/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. + * + *

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 { + + @Override + RecordWrapper wrapRecord(Record record) { + return new GenericRecordWrapper(record.getValue()); + } +} diff --git a/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/CassandraJsonStringSink.java b/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/CassandraJsonStringSink.java new file mode 100644 index 0000000000..1350bbbf88 --- /dev/null +++ b/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. + * + *

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. + * + *

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 { + + @Override + RecordWrapper wrapRecord(Record record) { + return new StringRecordWrapper(record.getValue()); + } +} diff --git a/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraSinkConfig.java b/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/CassandraSinkConfig.java similarity index 84% rename from cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraSinkConfig.java rename to cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/CassandraSinkConfig.java index b83af7a4c8..eade6db2d7 100644 --- a/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraSinkConfig.java +++ b/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/CassandraSinkConfig.java @@ -60,10 +60,15 @@ 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, @@ -71,9 +76,11 @@ public class CassandraSinkConfig implements Serializable { 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 { diff --git a/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/CassandraTableSink.java b/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/CassandraTableSink.java new file mode 100644 index 0000000000..dc3d00897a --- /dev/null +++ b/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. + * + *

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}. + * + *

{@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 implements Sink { + + CassandraConnector connector; + CassandraSinkConfig cassandraSinkConfig; + PreparedStatement stmt; + BoundStatementProvider boundStatementProvider; + + @Override + public void open(Map 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 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() { + @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 wrapRecord(Record record); + + PreparedStatement getStatement() { + if (stmt == null) { + stmt = connector.getPreparedStatement(); + } + return stmt; + } +} diff --git a/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/util/BoundStatementProvider.java b/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/util/BoundStatementProvider.java new file mode 100644 index 0000000000..db8b9f12a9 --- /dev/null +++ b/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; + } + +} diff --git a/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/util/CassandraConnector.java b/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/util/CassandraConnector.java new file mode 100644 index 0000000000..3e1609a80a --- /dev/null +++ b/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/util/CassandraConnector.java @@ -0,0 +1,147 @@ +/* + * 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.Cluster; +import com.datastax.driver.core.ColumnMetadata; +import com.datastax.driver.core.Metadata; +import com.datastax.driver.core.PreparedStatement; +import com.datastax.driver.core.Session; +import com.datastax.driver.core.TableMetadata; +import java.util.ArrayList; +import java.util.List; +import org.apache.pulsar.io.cassandra.CassandraSinkConfig; + +public class CassandraConnector implements AutoCloseable { + + private Cluster cluster; + private Session session; + private PreparedStatement statement; + private List tableFields; + private final CassandraSinkConfig config; + + public CassandraConnector(CassandraSinkConfig config) { + this.config = config; + } + + public void connect() { + session = getCluster().connect(config.getKeyspace()); + } + + public synchronized Session getSession() { + if (session == null || session.isClosed()) { + this.connect(); + } + return session; + } + + public Metadata getTableMetadata() { + return getCluster().getMetadata(); + } + + public PreparedStatement getPreparedStatement() { + if (statement == null) { + List fields = getTableFields(); + + StringBuilder sb = new StringBuilder("INSERT INTO ") + .append(config.getKeyspace() + "." + config.getColumnFamily() + " ("); + + for (int idx = 0; idx < fields.size(); idx++) { + sb.append(fields.get(idx)); + if (idx < fields.size() - 1) { + sb.append(", "); + } + } + + sb.append(") VALUES ("); + + for (int idx = 0; idx < fields.size(); idx++) { + sb.append("?"); + if (idx < fields.size() - 1) { + sb.append(", "); + } + } + + sb.append(")"); + statement = getSession().prepare(sb.toString()); + } + + return statement; + } + + List getTableFields() { + + if (tableFields == null) { + // Resolved through TableMetadataProvider so the column list backing the INSERT is the same + // one the binder positions values against, and so a bad keyspace/table names the setting + // that is wrong rather than surfacing as a NullPointerException. + TableMetadataProvider.TableDefinition table = TableMetadataProvider.getTableDefinition( + getCluster().getMetadata(), config.getKeyspace(), config.getColumnFamily()); + + tableFields = new ArrayList(table.getColumns().size()); + + for (TableMetadataProvider.ColumnId col : table.getColumns()) { + tableFields.add(col.getName()); + } + } + return tableFields; + } + + private synchronized Cluster getCluster() { + if (cluster == null) { + String[] hosts = config.getRoots().split(","); + + Cluster.Builder builder = Cluster.builder().withoutMetrics(); + + for (int i = 0; i < hosts.length; ++i) { + String[] hostPort = hosts[i].split(":"); + builder.addContactPoint(hostPort[0]); + if (hostPort.length > 1) { + builder.withPort(Integer.parseInt(hostPort[1])); + } + } + + // Authenticate if credentials have been provided. The sink has already rejected a pair + // with only one half set, so this is asking whether to authenticate, not whether the + // configuration makes sense. + if (config.hasCredentials()) { + builder.withCredentials( + config.getUserName(), + config.getPassword() + ); + } + cluster = builder.build(); + } + + return cluster; + } + + public void close() { + // Guarded on the fields, not on getSession()/getCluster(): those create what they return, so + // closing after a failed open() would re-attempt the connection and throw a second exception + // out of close(), hiding the original cause — or succeed and leave behind a session this method + // has just opened in order to close it. + if (session != null && !session.isClosed()) { + session.close(); + } + if (cluster != null && !cluster.isClosed()) { + cluster.close(); + } + } +} diff --git a/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/util/GenericRecordWrapper.java b/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/util/GenericRecordWrapper.java new file mode 100644 index 0000000000..d177d9490f --- /dev/null +++ b/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/util/GenericRecordWrapper.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.util; + +import org.apache.pulsar.client.api.schema.Field; +import org.apache.pulsar.client.api.schema.GenericRecord; + +public class GenericRecordWrapper extends RecordWrapper { + + public GenericRecordWrapper(GenericRecord value) { + super(value); + } + + @Override + public Object get(TableMetadataProvider.ColumnId column) { + return getValueAsExpectedType(recordValue.getField(column.getName()), column); + } + + @Override + public boolean containsKey(String name) { + // Ask the schema rather than the value. GenericAvroRecord.getField(name) delegates to Avro, + // which throws AvroRuntimeException for a name the schema does not carry instead of returning + // null. A table column the record has no field for is the ordinary case here — the table + // decides the column list, not the record — so this has to answer false, not fail the write. + // Iterated rather than streamed: this is asked once per table column for every record + // written, and a record's field list is short enough that the stream would cost more than + // the scan. + for (Field field : recordValue.getFields()) { + if (field.getName().equals(name)) { + return true; + } + } + return false; + } +} diff --git a/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/util/RecordWrapper.java b/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/util/RecordWrapper.java new file mode 100644 index 0000000000..f9c7e6eb6c --- /dev/null +++ b/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/util/RecordWrapper.java @@ -0,0 +1,88 @@ +/* + * 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.DataType; +import java.util.EnumSet; +import java.util.Locale; +import java.util.stream.Collectors; +import org.apache.commons.beanutils.converters.IntegerConverter; +import org.apache.commons.beanutils.converters.NumberConverter; + +public abstract class RecordWrapper { + + T recordValue; + NumberConverter converter = new IntegerConverter(); + + public RecordWrapper(T value) { + this.recordValue = value; + } + + public abstract Object get(TableMetadataProvider.ColumnId column); + + public abstract boolean containsKey(String name); + + /** + * Column types this wrapper can reliably bind a JSON or Avro value to. The set is deliberately + * narrow. Types such as {@code timestamp}, {@code uuid}, {@code inet}, {@code decimal} and + * {@code varint} need driver-specific Java objects ({@code Date}, {@code UUID}, + * {@code InetAddress}, {@code BigDecimal}, {@code BigInteger}) that neither Jackson nor Avro ever + * produces, and {@code bigint}/{@code smallint}/{@code tinyint} are worse than unsupported — they + * work or fail depending on the magnitude of the value, since Jackson decodes a small number as an + * {@code Integer}. Rejecting the table at {@code open()} beats an {@code InvalidTypeException} on + * every record, or on some records. + */ + private static final EnumSet SUPPORTED_COLUMN_TYPES = EnumSet.of( + DataType.Name.TEXT, + DataType.Name.VARCHAR, + DataType.Name.ASCII, + DataType.Name.INT, + DataType.Name.DOUBLE, + DataType.Name.FLOAT, + DataType.Name.BOOLEAN); + + public static boolean supports(DataType type) { + return SUPPORTED_COLUMN_TYPES.contains(type.getName()); + } + + public static String supportedColumnTypes() { + return SUPPORTED_COLUMN_TYPES.stream() + .map(name -> name.toString().toLowerCase(Locale.ROOT)) + .collect(Collectors.joining(", ")); + } + + Object getValueAsExpectedType(Object value, TableMetadataProvider.ColumnId column) { + // A field that is present but null binds as null. Every branch below would otherwise fail on + // it: TEXT calls toString(), and the converters reject a null with no default configured. + if (value == null) { + return null; + } + switch (column.getType().getName()) { + case FLOAT: return converter.convert(Float.class, value); + case INT: return converter.convert(Integer.class, value); + case DOUBLE: return converter.convert(Double.class, value); + case TEXT: + case VARCHAR: + case ASCII: return value.toString(); + default: return value; + } + + } + +} diff --git a/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/util/StringRecordWrapper.java b/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/util/StringRecordWrapper.java new file mode 100644 index 0000000000..45c3c50b7c --- /dev/null +++ b/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/util/StringRecordWrapper.java @@ -0,0 +1,52 @@ +/* + * 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.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.Map; + +public class StringRecordWrapper extends RecordWrapper { + private static final ObjectMapper MAPPER = new ObjectMapper(); + private static final TypeReference> MAP_TYPE = + new TypeReference>() { }; + + private final Map valuesMap; + + public StringRecordWrapper(String jsonString) { + super(jsonString); + try { + valuesMap = MAPPER.readValue(jsonString, MAP_TYPE); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + + @Override + public Object get(TableMetadataProvider.ColumnId column) { + return getValueAsExpectedType(valuesMap.get(column.getName()), column); + } + + @Override + public boolean containsKey(String name) { + return valuesMap.containsKey(name); + } +} diff --git a/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/util/TableMetadataProvider.java b/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/util/TableMetadataProvider.java new file mode 100644 index 0000000000..52c5f46b71 --- /dev/null +++ b/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/util/TableMetadataProvider.java @@ -0,0 +1,109 @@ +/* + * 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.ColumnMetadata; +import com.datastax.driver.core.DataType; +import com.datastax.driver.core.KeyspaceMetadata; +import com.datastax.driver.core.Metadata; +import com.datastax.driver.core.TableMetadata; +import com.google.common.collect.Lists; +import java.util.List; +import lombok.Data; +import lombok.EqualsAndHashCode; +import lombok.Getter; +import lombok.Setter; +import lombok.ToString; + +public class TableMetadataProvider { + + @Data(staticConstructor = "of") + public static class TableId { + private final String keyspaceName; + private final String columnFamily; + } + + @Data(staticConstructor = "of") + public static class ColumnId { + private final TableId tableId; + private final String name; + private final DataType type; + } + + @Setter + @Getter + @EqualsAndHashCode + @ToString + public static class TableDefinition { + private final TableId tableId; + private final List columns; + private final List partitionKeyColumns; + private final List primaryKeyColumns; + + private TableDefinition(TableId tableId, List columns, + List partitionKeyColumns, List primaryKeyColumns) { + this.tableId = tableId; + this.columns = columns; + this.partitionKeyColumns = partitionKeyColumns; + this.primaryKeyColumns = primaryKeyColumns; + } + + public static TableDefinition of(TableId tableId, List columns, + List partitionKeyColumns, + List primaryKeyColumns) { + return new TableDefinition(tableId, columns, partitionKeyColumns, primaryKeyColumns); + } + + } + + public static TableDefinition getTableDefinition(Metadata clusterMetadata, String keyspace, String columnFamily) { + + TableId tableId = TableId.of(keyspace, columnFamily); + TableDefinition table = TableDefinition.of(tableId, + Lists.newArrayList(), Lists.newArrayList(), Lists.newArrayList()); + + KeyspaceMetadata keyspaceMeta = clusterMetadata.getKeyspace(keyspace); + if (keyspaceMeta == null) { + throw new IllegalArgumentException("Keyspace '" + keyspace + + "' does not exist on this cluster; check the keyspace setting."); + } + TableMetadata meta = keyspaceMeta.getTable(columnFamily); + if (meta == null) { + throw new IllegalArgumentException("Table '" + columnFamily + "' does not exist in keyspace '" + + keyspace + "'; check the columnFamily setting."); + } + + for (ColumnMetadata col : meta.getPrimaryKey()) { + ColumnId columnId = ColumnId.of(tableId, col.getName(), col.getType()); + table.getPrimaryKeyColumns().add(columnId); + } + + for (ColumnMetadata col : meta.getPartitionKey()) { + ColumnId columnId = ColumnId.of(tableId, col.getName(), col.getType()); + table.getPartitionKeyColumns().add(columnId); + } + + for (ColumnMetadata col : meta.getColumns()) { + ColumnId columnId = ColumnId.of(tableId, col.getName(), col.getType()); + table.getColumns().add(columnId); + } + + return table; + } +} diff --git a/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/util/package-info.java b/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/util/package-info.java new file mode 100644 index 0000000000..904579e812 --- /dev/null +++ b/cassandra-core/src/main/java/org/apache/pulsar/io/cassandra/util/package-info.java @@ -0,0 +1,19 @@ +/* + * 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; \ No newline at end of file diff --git a/cassandra-core/src/test/java/org/apache/pulsar/io/cassandra/CassandraTableSinkIntegrationTest.java b/cassandra-core/src/test/java/org/apache/pulsar/io/cassandra/CassandraTableSinkIntegrationTest.java new file mode 100644 index 0000000000..b3e6e9cf2f --- /dev/null +++ b/cassandra-core/src/test/java/org/apache/pulsar/io/cassandra/CassandraTableSinkIntegrationTest.java @@ -0,0 +1,326 @@ +/* + * 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.assertFalse; +import static org.testng.Assert.assertNotNull; +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.Row; +import com.datastax.driver.core.Session; +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import lombok.Data; +import org.apache.pulsar.client.api.schema.GenericRecord; +import org.apache.pulsar.client.api.schema.SchemaDefinition; +import org.apache.pulsar.client.impl.schema.AvroSchema; +import org.apache.pulsar.client.impl.schema.generic.GenericAvroSchema; +import org.apache.pulsar.functions.api.Record; +import org.apache.pulsar.io.cassandra.util.AbstractCassandraTest; +import org.apache.pulsar.io.core.Sink; +import org.apache.pulsar.io.core.SinkContext; +import org.testng.annotations.Test; + +/** + * Drives {@link CassandraGenericRecordSink} and {@link CassandraJsonStringSink} end to end — real + * sink, real cluster, real rows — where the rest of this module's coverage stops at the machinery + * underneath them or at {@code open()}. + * + *

What these sinks promise is that the target table decides what gets written: fields are matched + * to columns by name, a field the table has no column for is dropped, and a column the record has no + * field for is left alone. A single-column write cannot show any of that, so both tests write to + * {@code airquality.reading} from {@code init.cql} — seventeen columns, a compound primary key, and + * {@code text} / {@code int} / {@code double} / {@code float} among them — populating seven of them + * and asserting on both what landed and what did not. + * + *

The generic-record case builds its input the way the hbase and jdbc sink tests do: encode a + * POJO with {@link AvroSchema}, decode it with {@link GenericAvroSchema}. That matters more than the + * convenience — it is a real Avro-backed record, so {@code text} columns receive Avro's string type + * rather than a {@link String} a hand-written test double would have handed over, and the coercion + * in {@code RecordWrapper} is exercised as it would be in a deployment. + */ +public class CassandraTableSinkIntegrationTest extends AbstractCassandraTest { + + private static final String KEYSPACE = "airquality"; + private static final String TABLE = "reading"; + + /** + * A subset of {@code airquality.reading}'s columns. Field names are the column names: that + * matching is the behaviour under test, and Cassandra folds unquoted identifiers to lower case, + * so the underscores are the table's, not a style choice. + */ + @Data + public static class Reading { + private String reporting_area; + private String date_observed; + private int hour_observed; + private String readingid; + private double avg_ozone; + private float latitude; + private String state_code; + } + + @Test + public void genericRecordSinkMapsFieldsOntoColumnsByName() throws Exception { + Reading reading = newReading("generic-record-area"); + + AvroSchema schema = + AvroSchema.of(SchemaDefinition.builder().withPojo(Reading.class).build()); + GenericRecord value = new GenericAvroSchema(schema.getSchemaInfo()).decode(schema.encode(reading)); + + writeThrough(new CassandraGenericRecordSink(), value); + + assertRowMatches(reading); + } + + @Test + public void jsonStringSinkMapsFieldsOntoColumnsByName() throws Exception { + Reading reading = newReading("json-string-area"); + + String json = "{" + + "\"reporting_area\":\"" + reading.getReporting_area() + "\"," + + "\"date_observed\":\"" + reading.getDate_observed() + "\"," + + "\"hour_observed\":" + reading.getHour_observed() + "," + + "\"readingid\":\"" + reading.getReadingid() + "\"," + + "\"avg_ozone\":" + reading.getAvg_ozone() + "," + + "\"latitude\":" + reading.getLatitude() + "," + + "\"state_code\":\"" + reading.getState_code() + "\"," + // No column of this name: the sink writes what the table has, and drops the rest. + + "\"not_a_column\":\"ignored\"" + + "}"; + + writeThrough(new CassandraJsonStringSink(), json); + + assertRowMatches(reading); + } + + /** + * A field that is present but null is a column written as null, not a failed write. Distinct from + * the {@code not_a_column} case above: there the field has no column, here the column has no + * value, and the two take different paths through {@code RecordWrapper}. + */ + @Test + public void jsonStringSinkWritesAnExplicitlyNullFieldAsNull() throws Exception { + Reading reading = newReading("json-null-area"); + reading.setState_code(null); + + String json = "{" + + "\"reporting_area\":\"" + reading.getReporting_area() + "\"," + + "\"date_observed\":\"" + reading.getDate_observed() + "\"," + + "\"hour_observed\":" + reading.getHour_observed() + "," + + "\"readingid\":\"" + reading.getReadingid() + "\"," + + "\"avg_ozone\":" + reading.getAvg_ozone() + "," + + "\"latitude\":" + reading.getLatitude() + "," + + "\"state_code\":null" + + "}"; + + writeThrough(new CassandraJsonStringSink(), json); + + assertRowMatches(reading); + } + + /** + * A second write to the same primary key that omits a field must leave that column as it was, not + * erase it. This is the promise the other tests describe but cannot check: on a fresh row an + * absent column and a column written as null are indistinguishable, and it takes a re-write to + * tell them apart. Binding an {@code Object[]} through {@code bind(Object...)} puts a real null in + * every unmatched position, and in Cassandra a null is a deletion, so this failed before + * {@link org.apache.pulsar.io.cassandra.util.BoundStatementProvider} began unsetting them. + */ + @Test + public void writingSameKeyAgainLeavesUnmatchedColumnsIntact() throws Exception { + Reading first = newReading("rewrite-area"); + + writeThrough(new CassandraJsonStringSink(), "{" + + "\"reporting_area\":\"" + first.getReporting_area() + "\"," + + "\"date_observed\":\"" + first.getDate_observed() + "\"," + + "\"hour_observed\":" + first.getHour_observed() + "," + + "\"state_code\":\"" + first.getState_code() + "\"," + + "\"readingid\":\"" + first.getReadingid() + "\"" + + "}"); + + // Same primary key, and this time state_code is not mentioned at all. + writeThrough(new CassandraJsonStringSink(), "{" + + "\"reporting_area\":\"" + first.getReporting_area() + "\"," + + "\"date_observed\":\"" + first.getDate_observed() + "\"," + + "\"hour_observed\":" + first.getHour_observed() + "," + + "\"readingid\":\"second-write\"" + + "}"); + + try (Cluster cluster = cassandraContainer.getCluster(); + Session session = cluster.connect(KEYSPACE)) { + Row row = session.execute("SELECT * FROM " + TABLE + " WHERE reporting_area = '" + + first.getReporting_area() + "'").one(); + assertNotNull(row); + assertEquals(row.getString("readingid"), "second-write", "The second write should have landed"); + assertEquals(row.getString("state_code"), first.getState_code(), + "A column the second record never mentioned was erased"); + } + } + + /** + * A record whose content cannot be bound — malformed JSON here — must be failed, not thrown out of + * {@code write()}. An escaping exception leaves the record neither acked nor failed, which kills + * the sink and has Pulsar redeliver the same message forever. + */ + @Test + public void unbindableRecordIsFailedRatherThanThrown() throws Exception { + CassandraJsonStringSink sink = new CassandraJsonStringSink(); + CompletableFuture acked = new CompletableFuture<>(); + AtomicBoolean failed = new AtomicBoolean(); + try { + sink.open(sinkConfig(), mock(SinkContext.class)); + sink.write(new Record() { + @Override + public Optional getKey() { + return Optional.empty(); + } + + @Override + public String getValue() { + return "this is not json"; + } + + @Override + public void ack() { + acked.complete(null); + } + + @Override + public void fail() { + failed.set(true); + } + }); + } finally { + sink.close(); + } + + assertTrue(failed.get(), "The record should have been failed"); + assertFalse(acked.isDone(), "The record should not have been acked"); + } + + /** + * A table holding a column type the sink cannot bind is refused at {@code open()}, naming the + * column and the type, rather than throwing an {@code InvalidTypeException} for every record it is + * later asked to write. + */ + @Test + public void openIsRefusedForAnUnsupportedColumnType() throws Exception { + Map config = sinkConfig(); + config.put("columnFamily", "unsupported_column_type"); + + CassandraJsonStringSink sink = new CassandraJsonStringSink(); + try { + sink.open(config, mock(SinkContext.class)); + fail("Expected open() to refuse a table with a timestamp column"); + } catch (IllegalArgumentException e) { + assertTrue(e.getMessage().contains("observed_at") && e.getMessage().contains("timestamp"), + "Message should name the offending column and type, got: " + e.getMessage()); + } finally { + sink.close(); + } + } + + private Reading newReading(String area) { + Reading reading = new Reading(); + reading.setReporting_area(area); + reading.setDate_observed("2026-08-17"); + reading.setHour_observed(9); + reading.setReadingid(area + "-1"); + reading.setAvg_ozone(0.031); + reading.setLatitude(37.77f); + reading.setState_code("CA"); + return reading; + } + + private void writeThrough(Sink sink, T value) throws Exception { + try { + sink.open(sinkConfig(), mock(SinkContext.class)); + + CompletableFuture acked = new CompletableFuture<>(); + sink.write(new Record() { + @Override + public Optional getKey() { + return Optional.empty(); + } + + @Override + public T getValue() { + return value; + } + + @Override + public void ack() { + acked.complete(null); + } + + @Override + public void fail() { + acked.completeExceptionally(new RuntimeException("Record failed")); + } + }); + acked.get(30, TimeUnit.SECONDS); + } finally { + sink.close(); + } + } + + private void assertRowMatches(Reading expected) { + try (Cluster cluster = cassandraContainer.getCluster(); + Session session = cluster.connect(KEYSPACE)) { + + Row row = session.execute("SELECT * FROM " + TABLE + " WHERE reporting_area = '" + + expected.getReporting_area() + "'").one(); + assertNotNull(row, "The sink acknowledged the record but no row was written"); + + assertEquals(row.getString("reporting_area"), expected.getReporting_area()); + assertEquals(row.getString("date_observed"), expected.getDate_observed()); + assertEquals(row.getInt("hour_observed"), expected.getHour_observed()); + assertEquals(row.getString("readingid"), expected.getReadingid()); + assertEquals(row.getString("state_code"), expected.getState_code()); + assertEquals(row.getDouble("avg_ozone"), expected.getAvg_ozone(), 0.000001); + assertEquals(row.getFloat("latitude"), expected.getLatitude(), 0.000001f); + + // Columns the record said nothing about are left alone rather than written as some + // default. Without this the assertions above would also pass for a sink that bound every + // column it could reach. + assertNull(row.getObject("max_ozone"), "A column with no matching field was written to"); + assertNull(row.getObject("local_time_zone"), "A column with no matching field was written to"); + } + } + + private Map sinkConfig() { + Map config = new HashMap<>(); + config.put("roots", cassandraContainer.getHost() + ":" + + cassandraContainer.getMappedPort(org.testcontainers.containers.CassandraContainer.CQL_PORT)); + config.put("keyspace", KEYSPACE); + config.put("columnFamily", TABLE); + config.put("userName", cassandraContainer.getUsername()); + config.put("password", cassandraContainer.getPassword()); + return config; + } +} diff --git a/cassandra-core/src/test/java/org/apache/pulsar/io/cassandra/util/AbstractCassandraTest.java b/cassandra-core/src/test/java/org/apache/pulsar/io/cassandra/util/AbstractCassandraTest.java new file mode 100644 index 0000000000..995f22fc73 --- /dev/null +++ b/cassandra-core/src/test/java/org/apache/pulsar/io/cassandra/util/AbstractCassandraTest.java @@ -0,0 +1,103 @@ +/* + * 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.Cluster; +import com.datastax.driver.core.Session; +import java.io.IOException; +import java.io.InputStream; +import java.io.UncheckedIOException; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Arrays; +import java.util.List; +import java.util.stream.Collectors; +import org.apache.pulsar.io.cassandra.CassandraSinkConfig; +import org.testcontainers.containers.CassandraContainer; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; + +/** + * Starts a Cassandra container once per test class and applies the {@code init.cql} schema that + * {@link CassandraConnectorTest} and {@link TableMetadataProviderTest} assert against. + * + *

The schema is applied through the Datastax driver rather than the container's own + * {@code withInitScript} support: the script delegate in the {@code org.testcontainers.cassandra} + * module is compiled against shaded classes that the Testcontainers core version resolved here no + * longer ships, so it fails at runtime. Going through the driver keeps this independent of that. + */ +public class AbstractCassandraTest { + + private static final String INIT_SCRIPT = "init.cql"; + + protected CassandraSinkConfig config; + protected CassandraContainer cassandraContainer; + + @BeforeClass + public void startCassandraContainer() { + cassandraContainer = new CassandraContainer<>("cassandra:4.1") + .withStartupTimeout(Duration.ofMinutes(3)); + cassandraContainer.start(); + applyInitScript(); + } + + @AfterClass(alwaysRun = true) + public void stopCassandraContainer() { + if (cassandraContainer != null) { + cassandraContainer.stop(); + cassandraContainer = null; + } + } + + protected void createSinkConfig() { + config = new CassandraSinkConfig(); + config.setRoots(cassandraContainer.getHost() + ":" + + cassandraContainer.getMappedPort(CassandraContainer.CQL_PORT)); + config.setUserName(cassandraContainer.getUsername()); + config.setPassword(cassandraContainer.getPassword()); + } + + private void applyInitScript() { + try (Cluster cluster = cassandraContainer.getCluster(); + Session session = cluster.connect()) { + for (String statement : readInitStatements()) { + session.execute(statement); + } + } + } + + private List readInitStatements() { + try (InputStream in = getClass().getClassLoader().getResourceAsStream(INIT_SCRIPT)) { + if (in == null) { + throw new IllegalStateException(INIT_SCRIPT + " not found on the test classpath"); + } + String script = new String(in.readAllBytes(), StandardCharsets.UTF_8); + // Strip line comments before splitting: a ';' never appears inside one in this script. + String stripped = Arrays.stream(script.split("\n")) + .filter(line -> !line.trim().startsWith("--")) + .collect(Collectors.joining("\n")); + return Arrays.stream(stripped.split(";")) + .map(String::trim) + .filter(s -> !s.isEmpty()) + .collect(Collectors.toList()); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } +} diff --git a/cassandra-core/src/test/java/org/apache/pulsar/io/cassandra/util/CassandraConnectorTest.java b/cassandra-core/src/test/java/org/apache/pulsar/io/cassandra/util/CassandraConnectorTest.java new file mode 100644 index 0000000000..079f112180 --- /dev/null +++ b/cassandra-core/src/test/java/org/apache/pulsar/io/cassandra/util/CassandraConnectorTest.java @@ -0,0 +1,64 @@ +/* + * 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 static org.testng.AssertJUnit.assertEquals; +import static org.testng.AssertJUnit.assertNotNull; +import lombok.Cleanup; +import org.testng.annotations.Test; + +public class CassandraConnectorTest extends AbstractCassandraTest { + + @Test + public final void securedTest() { + createSinkConfig(); + + @Cleanup + CassandraConnector connector = new CassandraConnector(config); + connector.connect(); + assertNotNull(connector.getSession()); + } + + @Test + public final void getObservationPreparedStatementTest() { + createSinkConfig(); + config.setColumnFamily("observation"); + config.setKeyspace("airquality"); + + @Cleanup + CassandraConnector connector = new CassandraConnector(config); + assertEquals("INSERT INTO airquality.observation (key, observed) VALUES (?, ?)", + connector.getPreparedStatement().getQueryString()); + } + + @Test + public final void getReadingPreparedStatementTest() { + createSinkConfig(); + config.setColumnFamily("reading"); + config.setKeyspace("airquality"); + + @Cleanup + CassandraConnector connector = new CassandraConnector(config); + assertEquals("INSERT INTO airquality.reading " + + "(reporting_area, date_observed, hour_observed, avg_ozone, avg_pm10, avg_pm25, latitude, " + + "local_time_zone, longitude, max_ozone, max_pm10, max_pm25, min_ozone, min_pm10, min_pm25, " + + "readingid, state_code) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + connector.getPreparedStatement().getQueryString()); + } +} diff --git a/cassandra-core/src/test/java/org/apache/pulsar/io/cassandra/util/TableMetadataProviderTest.java b/cassandra-core/src/test/java/org/apache/pulsar/io/cassandra/util/TableMetadataProviderTest.java new file mode 100644 index 0000000000..5d3c029234 --- /dev/null +++ b/cassandra-core/src/test/java/org/apache/pulsar/io/cassandra/util/TableMetadataProviderTest.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.util; + +import static org.testng.Assert.assertNotNull; +import static org.testng.AssertJUnit.assertEquals; +import lombok.Cleanup; +import org.testng.annotations.Test; + +public class TableMetadataProviderTest extends AbstractCassandraTest { + + @Test + public final void getTableDefinitionTest() { + + createSinkConfig(); + config.setColumnFamily("reading"); + config.setKeyspace("airquality"); + + @Cleanup + CassandraConnector connector = new CassandraConnector(config); + + TableMetadataProvider.TableDefinition table = + TableMetadataProvider.getTableDefinition( + connector.getTableMetadata(), + "airquality", "reading"); + + assertNotNull(table); + assertEquals(17, table.getColumns().size()); + assertEquals(1, table.getPartitionKeyColumns().size()); + assertEquals(3, table.getPrimaryKeyColumns().size()); + } +} diff --git a/cassandra-core/src/test/resources/init.cql b/cassandra-core/src/test/resources/init.cql new file mode 100644 index 0000000000..1ac81ea1c7 --- /dev/null +++ b/cassandra-core/src/test/resources/init.cql @@ -0,0 +1,67 @@ +-- 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. + +-- Create keyspace for air quality data +CREATE KEYSPACE IF NOT EXISTS airquality +WITH replication = { + 'class': 'SimpleStrategy', + 'replication_factor': 1 +}; + +-- Use the airquality keyspace +USE airquality; + +-- Create observation table +CREATE TABLE IF NOT EXISTS observation ( + key text PRIMARY KEY, + observed text +); + +-- Create reading table with air quality measurements +CREATE TABLE IF NOT EXISTS reading ( + reporting_area text, + date_observed text, + hour_observed int, + readingid text, + avg_ozone double, + min_ozone double, + max_ozone double, + avg_pm10 double, + min_pm10 double, + max_pm10 double, + avg_pm25 double, + min_pm25 double, + max_pm25 double, + local_time_zone text, + state_code text, + latitude float, + longitude float, + PRIMARY KEY ((reporting_area), date_observed, hour_observed) +) WITH CLUSTERING ORDER BY (date_observed DESC, hour_observed DESC); + +-- Create index on readingid for queries by reading ID +CREATE INDEX IF NOT EXISTS reading_id_idx ON reading (readingid); + +-- Create index on state_code for queries by state +CREATE INDEX IF NOT EXISTS state_code_idx ON reading (state_code); + +-- Holds a column type the table-mapping sinks cannot bind a record field onto, so that their +-- refusal to open against such a table can be asserted. +CREATE TABLE IF NOT EXISTS unsupported_column_type ( + key text PRIMARY KEY, + observed_at timestamp +); diff --git a/cassandra-generic-record/build.gradle.kts b/cassandra-generic-record/build.gradle.kts new file mode 100644 index 0000000000..540ef0da87 --- /dev/null +++ b/cassandra-generic-record/build.gradle.kts @@ -0,0 +1,29 @@ +/* + * 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") + id("pulsar-connectors.nar-conventions") +} +dependencies { + // The sink classes themselves live in :cassandra-core, shared with the `cassandra` sink; this + // module exists only to give this connector its own NAR, because a NAR's pulsar-io.yaml can + // declare exactly one sinkClass and so one NAR can offer exactly one connector by name. + implementation(project(":cassandra-core")) +} diff --git a/cassandra-generic-record/src/main/resources/META-INF/services/pulsar-io.yaml b/cassandra-generic-record/src/main/resources/META-INF/services/pulsar-io.yaml new file mode 100644 index 0000000000..ad42128b4b --- /dev/null +++ b/cassandra-generic-record/src/main/resources/META-INF/services/pulsar-io.yaml @@ -0,0 +1,23 @@ +# +# 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. +# + +name: cassandra-generic-record +description: Writes schema-carrying records into Cassandra, mapping record fields onto table columns by name +sinkClass: org.apache.pulsar.io.cassandra.CassandraGenericRecordSink +sinkConfigClass: org.apache.pulsar.io.cassandra.CassandraSinkConfig diff --git a/cassandra-json/build.gradle.kts b/cassandra-json/build.gradle.kts new file mode 100644 index 0000000000..540ef0da87 --- /dev/null +++ b/cassandra-json/build.gradle.kts @@ -0,0 +1,29 @@ +/* + * 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") + id("pulsar-connectors.nar-conventions") +} +dependencies { + // The sink classes themselves live in :cassandra-core, shared with the `cassandra` sink; this + // module exists only to give this connector its own NAR, because a NAR's pulsar-io.yaml can + // declare exactly one sinkClass and so one NAR can offer exactly one connector by name. + implementation(project(":cassandra-core")) +} diff --git a/cassandra-json/src/main/resources/META-INF/services/pulsar-io.yaml b/cassandra-json/src/main/resources/META-INF/services/pulsar-io.yaml new file mode 100644 index 0000000000..e3efa0b5c7 --- /dev/null +++ b/cassandra-json/src/main/resources/META-INF/services/pulsar-io.yaml @@ -0,0 +1,23 @@ +# +# 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. +# + +name: cassandra-json +description: Writes raw JSON string messages into Cassandra, mapping top-level JSON fields onto table columns by name +sinkClass: org.apache.pulsar.io.cassandra.CassandraJsonStringSink +sinkConfigClass: org.apache.pulsar.io.cassandra.CassandraSinkConfig diff --git a/cassandra/build.gradle.kts b/cassandra/build.gradle.kts index 5e89fef8ef..e2fad99f61 100644 --- a/cassandra/build.gradle.kts +++ b/cassandra/build.gradle.kts @@ -22,6 +22,10 @@ plugins { id("pulsar-connectors.nar-conventions") } dependencies { + // CassandraSinkConfig and the table-mapping sinks live in :cassandra-core so that the NAR modules + // for the other two connectors can depend on them; see that module's build file for why it is not + // a NAR itself. + implementation(project(":cassandra-core")) implementation(libs.pulsar.io.core) implementation(libs.pulsar.io.common) implementation(libs.jackson.databind) diff --git a/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkConfigValidationTest.java b/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkConfigValidationTest.java new file mode 100644 index 0000000000..15235f1134 --- /dev/null +++ b/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkConfigValidationTest.java @@ -0,0 +1,126 @@ +/* + * 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.assertTrue; +import static org.testng.Assert.fail; +import java.util.HashMap; +import java.util.Map; +import org.apache.pulsar.io.core.Sink; +import org.apache.pulsar.io.core.SinkContext; +import org.testng.annotations.Test; + +/** + * Covers what each sink accepts and rejects at {@code open()} before it reaches the cluster, where + * the three sinks share one {@link CassandraSinkConfig} but do not share one contract. + * + *

{@code keyname} and {@code columnName} are the divergence. They name the two columns the + * {@code cassandra} sink writes, and mean nothing to the table-mapping sinks, which read the column + * list from the table itself. Marking them {@code required = true} would have + * {@code IOConfigUtils.loadWithSecrets} reject a table-sink config that omits them — before the sink + * gets to say it does not need them — so they are optional on the config and enforced by + * {@link CassandraAbstractSink} for the sink that does need them. Both halves of that are asserted + * here, because either alone would be wrong. + * + *

No container: {@code roots} points at a port nothing listens on. A configuration that gets as + * far as failing to connect is one that passed validation, which is what makes the accepting cases + * below say anything. + */ +public class CassandraSinkConfigValidationTest { + + @Test + public void genericRecordSinkDoesNotRequireKeynameOrColumnName() { + assertValidationPasses(new CassandraGenericRecordSink(), tableSinkConfig()); + } + + @Test + public void jsonStringSinkDoesNotRequireKeynameOrColumnName() { + assertValidationPasses(new CassandraJsonStringSink(), tableSinkConfig()); + } + + @Test + public void stringSinkStillRequiresKeyname() { + Map config = tableSinkConfig(); + config.put("columnName", "value"); + + assertRejected(new CassandraStringSink(), config, "Required property not set."); + } + + @Test + public void stringSinkStillRequiresColumnName() { + Map config = tableSinkConfig(); + config.put("keyname", "key"); + + assertRejected(new CassandraStringSink(), config, "Required property not set."); + } + + @Test + public void genericRecordSinkRejectsUsernameWithoutPassword() { + Map config = tableSinkConfig(); + config.put("userName", "cassandra"); + + assertRejected(new CassandraGenericRecordSink(), config, + "userName and password must be supplied together"); + } + + @Test + public void jsonStringSinkRejectsPasswordWithoutUsername() { + Map config = tableSinkConfig(); + config.put("password", "cassandra"); + + assertRejected(new CassandraJsonStringSink(), config, + "userName and password must be supplied together"); + } + + private void assertValidationPasses(Sink sink, Map config) { + try { + sink.open(config, mock(SinkContext.class)); + fail("Expected the unreachable contact point to fail the connection"); + } catch (IllegalArgumentException e) { + fail("Rejected as invalid rather than reaching the cluster: " + e.getMessage()); + } catch (Exception e) { + // Anything else means validation passed and the driver got as far as the network. + } + } + + private void assertRejected(Sink sink, Map config, String expected) { + try { + sink.open(config, mock(SinkContext.class)); + fail("Expected open() to reject this configuration"); + } catch (IllegalArgumentException e) { + assertTrue(e.getMessage().contains(expected), + "Rejected for some other reason: " + e.getMessage()); + } catch (Exception e) { + fail("Expected IllegalArgumentException, got: " + e); + } + } + + /** + * The settings a table-mapping sink needs, and nothing else: no {@code keyname}, no + * {@code columnName}. Nothing listens on the port named here. + */ + private Map tableSinkConfig() { + Map config = new HashMap<>(); + config.put("roots", "127.0.0.1:1"); + config.put("keyspace", "any_keyspace"); + config.put("columnFamily", "any_table"); + return config; + } +} diff --git a/distribution/io/build.gradle.kts b/distribution/io/build.gradle.kts index 0ac5176e25..b8eb81b59d 100644 --- a/distribution/io/build.gradle.kts +++ b/distribution/io/build.gradle.kts @@ -38,6 +38,8 @@ val connectorNars by configurations.creating { dependencies { connectorNars(project(":cassandra")) + connectorNars(project(":cassandra-generic-record")) + connectorNars(project(":cassandra-json")) connectorNars(project(":kafka")) connectorNars(project(":http")) connectorNars(project(":kinesis")) diff --git a/docs/build.gradle.kts b/docs/build.gradle.kts index 7d645b1209..35ead05f95 100644 --- a/docs/build.gradle.kts +++ b/docs/build.gradle.kts @@ -29,6 +29,8 @@ dependencies { implementation(project(":aerospike")) implementation(project(":canal")) implementation(project(":cassandra")) + implementation(project(":cassandra-generic-record")) + implementation(project(":cassandra-json")) implementation(project(":debezium:pulsar-io-debezium-mariadb")) implementation(project(":debezium:pulsar-io-debezium-mysql")) implementation(project(":debezium:pulsar-io-debezium-postgres")) diff --git a/settings.gradle.kts b/settings.gradle.kts index c186ce52c5..99fd9b3ca5 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -69,6 +69,9 @@ include("aws") include("azure-data-explorer") include("canal") include("cassandra") +include("cassandra-core") +include("cassandra-generic-record") +include("cassandra-json") include("dynamodb") include("elastic-search") include("file")