DCC Schema Refresh Fix

pom.xml

@@ -4,20 +4,20 @@
     <parent>
         <groupId>io.debezium</groupId>
         <artifactId>debezium-parent</artifactId>
-        <version>1.5.0-SNAPSHOT</version>
+        <version>1.5.0.Final</version>
     </parent>
 
     <modelVersion>4.0.0</modelVersion>
     <artifactId>debezium-connector-cassandra</artifactId>
     <name>Debezium Connector for Cassandra</name>
-    <version>1.5.0-SNAPSHOT</version>
+    <version>1.5.0.Final</version>
     <packaging>jar</packaging>
 
     <scm>
         <connection>scm:git:[email protected]:debezium/debezium-connector-cassandra.git</connection>
         <developerConnection>scm:git:[email protected]:debezium/debezium-connector-cassandra.git</developerConnection>
         <url>https://github.com/debezium/debezium-connector-cassandra</url>
-      <tag>HEAD</tag>
+      <tag>v1.5.0.Final</tag>
     </scm>
 
     <properties>
@@ -119,6 +119,7 @@
         <dependency>
             <groupId>org.apache.cassandra</groupId>
             <artifactId>cassandra-all</artifactId>
+            <version>3.11.4</version>
             <exclusions>
                 <exclusion>
                     <groupId>ch.qos.logback</groupId>

src/main/java/io/debezium/connector/cassandra/CassandraClient.java

@@ -95,6 +95,10 @@ public String getClusterName() {
         return cluster.getMetadata().getClusterName();
     }
 
+    public Cluster getCluster() {
+        return cluster;
+    }
+
     public boolean isQueryable() {
         return !cluster.isClosed() && !session.isClosed();
     }

src/main/java/io/debezium/connector/cassandra/CassandraConnectorContext.java

@@ -53,7 +53,7 @@ public CassandraConnectorContext(CassandraConnectorConfig config) throws Excepti
                     .build();
 
             // Setting up schema holder ...
-            this.schemaHolder = new SchemaHolder(this.cassandraClient, this.config.kafkaTopicPrefix(), this.config.getSourceInfoStructMaker());
+            this.schemaHolder = new SchemaHolder(this.config.kafkaTopicPrefix(), this.config.getSourceInfoStructMaker());
 
             // Setting up a file-based offset manager ...
             this.offsetWriter = new FileOffsetWriter(this.config.offsetBackingStoreDir());

src/main/java/io/debezium/connector/cassandra/CommitLogReadHandlerImpl.java

@@ -48,7 +48,7 @@
 
 /**
  * Handler that implements {@link CommitLogReadHandler} interface provided by Cassandra source code.
- *
+ * <p>
  * This handler implementation processes each {@link Mutation} and invokes one of the registered partition handler
  * for each {@link PartitionUpdate} in the {@link Mutation} (a mutation could have multiple partitions if it is a batch update),
  * which in turn makes one or more record via the {@link RecordMaker} and enqueue the record into the {@link ChangeEventQueue}.
@@ -77,7 +77,7 @@ public class CommitLogReadHandlerImpl implements CommitLogReadHandler {
     }
 
     /**
-     *  A PartitionType represents the type of a PartitionUpdate.
+     * A PartitionType represents the type of a PartitionUpdate.
      */
     enum PartitionType {
         /**
@@ -86,7 +86,7 @@ enum PartitionType {
         PARTITION_KEY_ROW_DELETION,
 
         /**
-         *  a partition-level deletion where partition key + clustering key = primary key
+         * a partition-level deletion where partition key + clustering key = primary key
          */
         PARTITION_AND_CLUSTERING_KEY_ROW_DELETION,
 
@@ -147,7 +147,7 @@ public static boolean isPartitionDeletion(PartitionUpdate pu) {
     }
 
     /**
-     *  A RowType represents different types of {@link Row}-level modifications in a Cassandra table.
+     * A RowType represents different types of {@link Row}-level modifications in a Cassandra table.
      */
     enum RowType {
         /**
@@ -303,18 +303,22 @@ private void process(PartitionUpdate pu, OffsetPosition offsetPosition, Keyspace
      * Handle a valid deletion event resulted from a partition-level deletion by converting Cassandra representation
      * of this event into a {@link Record} object and queue the record to {@link ChangeEventQueue}. A valid deletion
      * event means a partition only has a single row, this implies there are no clustering keys.
-     *
+     * <p>
      * The steps are:
-     *      (1) Populate the "source" field for this event
-     *      (2) Fetch the cached key/value schemas from {@link SchemaHolder}
-     *      (3) Populate the "after" field for this event
-     *          a. populate partition columns
-     *          b. populate regular columns with null values
-     *      (4) Assemble a {@link Record} object from the populated data and queue the record
+     * (1) Populate the "source" field for this event
+     * (2) Fetch the cached key/value schemas from {@link SchemaHolder}
+     * (3) Populate the "after" field for this event
+     * a. populate partition columns
+     * b. populate regular columns with null values
+     * (4) Assemble a {@link Record} object from the populated data and queue the record
      */
     private void handlePartitionDeletion(PartitionUpdate pu, OffsetPosition offsetPosition, KeyspaceTable keyspaceTable) {
 
-        SchemaHolder.KeyValueSchema keyValueSchema = schemaHolder.getOrUpdateKeyValueSchema(keyspaceTable);
+        SchemaHolder.KeyValueSchema keyValueSchema = schemaHolder.getKeyValueSchema(keyspaceTable);
+        if (keyValueSchema == null) {
+            return;
+        }
+
         Schema keySchema = keyValueSchema.keySchema();
         Schema valueSchema = keyValueSchema.valueSchema();
 
@@ -349,27 +353,30 @@ private void handlePartitionDeletion(PartitionUpdate pu, OffsetPosition offsetPo
      * Handle a valid event resulted from a row-level modification by converting Cassandra representation of
      * this event into a {@link Record} object and queue the record to {@link ChangeEventQueue}. A valid event
      * implies this must be an insert, update, or delete.
-     *
+     * <p>
      * The steps are:
-     *      (1) Populate the "source" field for this event
-     *      (2) Fetch the cached key/value schemas from {@link SchemaHolder}
-     *      (3) Populate the "after" field for this event
-     *          a. populate partition columns
-     *          b. populate clustering columns
-     *          c. populate regular columns
-     *          d. for deletions, populate regular columns with null values
-     *      (4) Assemble a {@link Record} object from the populated data and queue the record
+     * (1) Populate the "source" field for this event
+     * (2) Fetch the cached key/value schemas from {@link SchemaHolder}
+     * (3) Populate the "after" field for this event
+     * a. populate partition columns
+     * b. populate clustering columns
+     * c. populate regular columns
+     * d. for deletions, populate regular columns with null values
+     * (4) Assemble a {@link Record} object from the populated data and queue the record
      */
     private void handleRowModifications(Row row, RowType rowType, PartitionUpdate pu, OffsetPosition offsetPosition, KeyspaceTable keyspaceTable) {
 
-        SchemaHolder.KeyValueSchema schema = schemaHolder.getOrUpdateKeyValueSchema(keyspaceTable);
-        Schema keySchema = schema.keySchema();
-        Schema valueSchema = schema.valueSchema();
+        SchemaHolder.KeyValueSchema keyValueSchema = schemaHolder.getKeyValueSchema(keyspaceTable);
+        if (keyValueSchema == null) {
+            return;
+        }
+        Schema keySchema = keyValueSchema.keySchema();
+        Schema valueSchema = keyValueSchema.valueSchema();
 
         RowData after = new RowData();
         populatePartitionColumns(after, pu);
         populateClusteringColumns(after, row, pu);
-        populateRegularColumns(after, row, rowType, schema);
+        populateRegularColumns(after, row, rowType, keyValueSchema);
 
         long ts = rowType == DELETE ? row.deletion().time().markedForDeleteAt() : pu.maxTimestamp();
 

src/main/java/io/debezium/connector/cassandra/NoOpSchemaChangeListener.java

@@ -0,0 +1,117 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.connector.cassandra;
+
+import com.datastax.driver.core.AggregateMetadata;
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.FunctionMetadata;
+import com.datastax.driver.core.KeyspaceMetadata;
+import com.datastax.driver.core.MaterializedViewMetadata;
+import com.datastax.driver.core.SchemaChangeListener;
+import com.datastax.driver.core.TableMetadata;
+import com.datastax.driver.core.UserType;
+
+public class NoOpSchemaChangeListener implements SchemaChangeListener {
+    @Override
+    public void onKeyspaceAdded(final KeyspaceMetadata keyspace) {
+
+    }
+
+    @Override
+    public void onKeyspaceRemoved(final KeyspaceMetadata keyspace) {
+
+    }
+
+    @Override
+    public void onKeyspaceChanged(final KeyspaceMetadata current, final KeyspaceMetadata previous) {
+
+    }
+
+    @Override
+    public void onTableAdded(final TableMetadata table) {
+
+    }
+
+    @Override
+    public void onTableRemoved(final TableMetadata table) {
+
+    }
+
+    @Override
+    public void onTableChanged(final TableMetadata current, final TableMetadata previous) {
+
+    }
+
+    @Override
+    public void onUserTypeAdded(final UserType type) {
+
+    }
+
+    @Override
+    public void onUserTypeRemoved(final UserType type) {
+
+    }
+
+    @Override
+    public void onUserTypeChanged(final UserType current, final UserType previous) {
+
+    }
+
+    @Override
+    public void onFunctionAdded(final FunctionMetadata function) {
+
+    }
+
+    @Override
+    public void onFunctionRemoved(final FunctionMetadata function) {
+
+    }
+
+    @Override
+    public void onFunctionChanged(final FunctionMetadata current, final FunctionMetadata previous) {
+
+    }
+
+    @Override
+    public void onAggregateAdded(final AggregateMetadata aggregate) {
+
+    }
+
+    @Override
+    public void onAggregateRemoved(final AggregateMetadata aggregate) {
+
+    }
+
+    @Override
+    public void onAggregateChanged(final AggregateMetadata current, final AggregateMetadata previous) {
+
+    }
+
+    @Override
+    public void onMaterializedViewAdded(final MaterializedViewMetadata view) {
+
+    }
+
+    @Override
+    public void onMaterializedViewRemoved(final MaterializedViewMetadata view) {
+
+    }
+
+    @Override
+    public void onMaterializedViewChanged(final MaterializedViewMetadata current, final MaterializedViewMetadata previous) {
+
+    }
+
+    @Override
+    public void onRegister(final Cluster cluster) {
+
+    }
+
+    @Override
+    public void onUnregister(final Cluster cluster) {
+
+    }
+}