Adding option to push only post image events to kafka

Author: Chirag Bansal

README.md

@@ -15,20 +15,25 @@ The connector has the following capabilities:
     - `UPDATE`
     - `DELETE` (single row deletes)
 - High scalability - able to split work accross multiple Kafka Connect workers
-- Fault tolerant - connector periodically saves its progress and can resume from previously saved offset (with at-least-once semantics)
+- Fault-tolerant - connector periodically saves its progress and can resume from previously saved offset (with at-least-once semantics)
 - Support for many standard Kafka Connect converters, such as JSON and Avro
 - Compatible with standard Kafka Connect transformations
 - Metadata about CDC events - each generated Kafka message contains information about source, such as timestamp and table name
 - Seamless handling of schema changes and topology changes (adding, removing nodes from Scylla cluster)
 - Preimage support ([optional](#advanced-configuration-parameters)) - messages generated for row-level changes can have their [`before`](#data-change-event-value) field filled with information from corresponding preimage row.
+- Post image only: You can configure the connector to produce only `POST_IMAGE` cdc events as `CREATE` events. 
 
 The connector has the following limitations:
 - Only Kafka 2.6.0+ is supported
 - Only row-level operations are produced (`INSERT`, `UPDATE`, `DELETE`):
     - Partition deletes - those changes are ignored
     - Row range deletes - those changes are ignored
 - No support for collection types (`LIST`, `SET`, `MAP`) and `UDT` - columns with those types are omitted from generated messages
+<<<<<<< HEAD
 - No support for postimage, preimage needs to be enabled - By default changes only contain those columns that were modified, not the entire row before/after change. More information [here](#cell-representation)
+=======
+- No support for preimage - changes only contain those columns that were modified, not the entire row before/after change. More information [here](#cell-representation)
+>>>>>>> 4c29e07 (Adding option to push only post image events to kafka)
 
 ## Connector installation
 
@@ -684,6 +689,7 @@ The connector will generate the following data change event's value (with JSON s
 
 In addition to the configuration parameters described in the ["Configuration"](#configuration) section, Scylla CDC Source Connector exposes the following (non-required) configuration parameters:
 
+<<<<<<< HEAD
 | Property                         | Description                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                           |
 |----------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
 | `scylla.query.time.window.size`  | The size of windows queried by the connector. Changes are queried using `SELECT` statements with time restriction with width defined by this parameter. Value expressed in milliseconds.                                                                                                                                                                                                                                                                                                                              |
@@ -692,6 +698,15 @@ In addition to the configuration parameters described in the ["Configuration"](#
 | `scylla.local.dc`                | The name of Scylla local datacenter. This local datacenter name will be used to setup the connection to Scylla to prioritize sending requests to the nodes in the local datacenter. If not set, no particular datacenter will be prioritized.                                                                                                                                                                                                                                                                         |
 | `experimental.preimages.enabled` | False by default. If enabled connector will use `PRE_IMAGE` CDC entries to populate 'before' field of the debezium Envelope of the next kafka message. This may change some expected behaviours (e.g. ROW_DELETE will use preimage instead of its own information). Relies on correct ordering of rows within same stream in CDC tables.                                                                                                                                                                              |
 
+=======
+| Property | Description |
+| --- | --- |
+| `scylla.query.time.window.size` | The size of windows queried by the connector. Changes are queried using `SELECT` statements with time restriction with width defined by this parameter. Value expressed in milliseconds. |
+| `scylla.confidence.window.size` | The size of the confidence window. It is necessary for the connector to avoid reading too fresh data from the CDC log due to the eventual consistency of Scylla. The problem could appear when a newer write reaches a replica before some older write. For a short period of time, when reading, it is possible for the replica to return only the newer write. The connector mitigates this problem by not reading a window of most recent changes (controlled by this parameter). Value expressed in milliseconds.|
+| `scylla.consistency.level` | The consistency level of CDC table read queries. This consistency level is used only for read queries to the CDC log table. By default, `QUORUM` level is used. |
+| `scylla.local.dc` | The name of Scylla local datacenter. This local datacenter name will be used to setup the connection to Scylla to prioritize sending requests to the nodes in the local datacenter. If not set, no particular datacenter will be prioritized. |
+| `post.image.only` | Push only the post image events from scylla cdc to kafka. The events are pushed as `CREATE` events. |
+>>>>>>> 4c29e07 (Adding option to push only post image events to kafka)
 
 ### Configuration for large Scylla clusters
 #### Offset (progress) storage

src/main/java/com/scylladb/cdc/debezium/connector/ScyllaChangeRecordEmitter.java

@@ -51,6 +51,7 @@ protected Envelope.Operation getOperation() {
             case ROW_UPDATE:
                 return Envelope.Operation.UPDATE;
             case ROW_INSERT:
+            case POST_IMAGE:
                 return Envelope.Operation.CREATE;
             case PARTITION_DELETE: // See comment in ScyllaChangesConsumer on the support of partition deletes.
             case ROW_DELETE:

src/main/java/com/scylladb/cdc/debezium/connector/ScyllaChangesConsumer.java

@@ -3,6 +3,7 @@
 import com.scylladb.cdc.model.TaskId;
 import com.scylladb.cdc.model.worker.ChangeSchema;
 import com.scylladb.cdc.model.worker.RawChange;
+import com.scylladb.cdc.model.worker.RawChange.OperationType;
 import com.scylladb.cdc.model.worker.Task;
 import com.scylladb.cdc.model.worker.TaskAndRawChangeConsumer;
 import io.debezium.pipeline.EventDispatcher;
@@ -23,8 +24,9 @@ public class ScyllaChangesConsumer implements TaskAndRawChangeConsumer {
     private final Clock clock;
     private final boolean usePreimages;
     private final Map<TaskId, RawChange> lastPreImage;
+    private final boolean postImageOnly;
 
-    public ScyllaChangesConsumer(EventDispatcher<CollectionId> dispatcher, ScyllaOffsetContext offsetContext, ScyllaSchema schema, Clock clock, boolean usePreimages) {
+    public ScyllaChangesConsumer(EventDispatcher<CollectionId> dispatcher, ScyllaOffsetContext offsetContext, ScyllaSchema schema, Clock clock, boolean usePreimages, boolean postImageOnly) {
         this.dispatcher = dispatcher;
         this.offsetContext = offsetContext;
         this.schema = schema;
@@ -35,6 +37,7 @@ public ScyllaChangesConsumer(EventDispatcher<CollectionId> dispatcher, ScyllaOff
         } else {
             this.lastPreImage = null;
         }
+        this.postImageOnly = postImageOnly;
     }
 
     @Override
@@ -66,10 +69,13 @@ public CompletableFuture<Void> consume(Task task, RawChange change) {
                 if (hasClusteringColumn) {
                     return CompletableFuture.completedFuture(null);
                 }
-            } else if (operationType != RawChange.OperationType.ROW_INSERT
+            } else if (!this.postImageOnly
+                    && operationType != RawChange.OperationType.ROW_INSERT
                     && operationType != RawChange.OperationType.ROW_UPDATE
                     && operationType != RawChange.OperationType.ROW_DELETE) {
                 return CompletableFuture.completedFuture(null);
+            } else if (this.postImageOnly && operationType != OperationType.POST_IMAGE) {
+                return CompletableFuture.completedFuture(null);
             }
 
             if (usePreimages && lastPreImage.containsKey(task.id)) {

src/main/java/com/scylladb/cdc/debezium/connector/ScyllaConnectorConfig.java

@@ -167,6 +167,14 @@ public class ScyllaConnectorConfig extends CommonConnectorConfig {
             "driver's QueryOptions before session construction. Set this to an explicit value if " +
             "experiencing too high memory usage.");
 
+    public static final boolean POST_IMAGE_ONLY_DEFAULT = false;
+    public static final Field POST_IMAGE_ONLY = Field.create("post.image.only")
+            .withDisplayName("Post Image Only")
+            .withType(ConfigDef.Type.BOOLEAN)
+            .withWidth(ConfigDef.Width.SHORT)
+            .withImportance(ConfigDef.Importance.MEDIUM)
+            .withDescription("Whether the connector should publish only post image events. The cdc settings must have `'postimage': 'true'`. To get full image with all fields, set `'preimage': 'full'`.");
+
     public static final Field LOCAL_DC_NAME = Field.create("scylla.local.dc")
             .withDisplayName("Local DC Name")
             .withType(ConfigDef.Type.STRING)
@@ -205,7 +213,7 @@ public class ScyllaConnectorConfig extends CommonConnectorConfig {
                     .name("Scylla")
                     .type(CLUSTER_IP_ADDRESSES, USER, PASSWORD, LOGICAL_NAME, CONSISTENCY_LEVEL, QUERY_OPTIONS_FETCH_SIZE, LOCAL_DC_NAME, SSL_ENABLED, SSL_PROVIDER, SSL_TRUSTSTORE_PATH, SSL_TRUSTSTORE_PASSWORD, SSL_KEYSTORE_PATH, SSL_KEYSTORE_PASSWORD,SSL_CIPHER_SUITES, SSL_OPENSLL_KEYCERTCHAIN, SSL_OPENSLL_PRIVATEKEY)
                     .connector(QUERY_TIME_WINDOW_SIZE, CONFIDENCE_WINDOW_SIZE, PREIMAGES_ENABLED)
-                    .events(TABLE_NAMES)
+                    .events(TABLE_NAMES, POST_IMAGE_ONLY)
                     .excluding(Heartbeat.HEARTBEAT_INTERVAL).events(CUSTOM_HEARTBEAT_INTERVAL)
                     // Exclude some Debezium options, which are not applicable/not supported by
                     // the Scylla CDC Source Connector.
@@ -305,6 +313,10 @@ public CQLConfiguration.ConsistencyLevel getConsistencyLevel() {
         }
     }
 
+    public boolean isPostImageOnly() {
+        return config.getBoolean(ScyllaConnectorConfig.POST_IMAGE_ONLY, POST_IMAGE_ONLY_DEFAULT);
+    }
+
     public String getLocalDCName() {
         return config.getString(ScyllaConnectorConfig.LOCAL_DC_NAME);
     }

src/main/java/com/scylladb/cdc/debezium/connector/ScyllaStreamingChangeEventSource.java

@@ -2,20 +2,14 @@
 
 import com.scylladb.cdc.cql.driver3.Driver3Session;
 import com.scylladb.cdc.cql.driver3.Driver3WorkerCQL;
-import com.scylladb.cdc.model.ExponentialRetryBackoffWithJitter;
-import com.scylladb.cdc.model.RetryBackoff;
 import com.scylladb.cdc.model.worker.WorkerConfiguration;
 import com.scylladb.cdc.model.worker.Worker;
 import io.debezium.pipeline.EventDispatcher;
 import io.debezium.pipeline.source.spi.StreamingChangeEventSource;
 import io.debezium.util.Clock;
 import org.apache.commons.lang3.tuple.Pair;
 
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-import java.sql.Driver;
 import java.time.Duration;
-import java.util.List;
 import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
 
@@ -43,7 +37,7 @@ public void execute(ChangeEventSourceContext context) throws InterruptedExceptio
         Driver3Session session = new ScyllaSessionBuilder(configuration).build();
         Driver3WorkerCQL cql = new Driver3WorkerCQL(session);
         ScyllaWorkerTransport workerTransport = new ScyllaWorkerTransport(context, offsetContext, dispatcher, configuration.getHeartbeatIntervalMs());
-        ScyllaChangesConsumer changeConsumer = new ScyllaChangesConsumer(dispatcher, offsetContext, schema, clock, configuration.getPreimagesEnabled());
+        ScyllaChangesConsumer changeConsumer = new ScyllaChangesConsumer(dispatcher, offsetContext, schema, clock, configuration.getPreimagesEnabled(), configuration.isPostImageOnly());
         WorkerConfiguration workerConfiguration = WorkerConfiguration.builder()
                 .withTransport(workerTransport)
                 .withCQL(cql)