DBZ-5410 Added ability to process Cassandra 4 Commit log files incrementally #69

nitinitt · 2022-08-19T18:07:38Z

I had a merge commit in my previous PR: #62 and am not able to remove that without altering the git history of "main" branch. Please accept this PR as part of:
https://issues.redhat.com/browse/DBZ-5410

nitinitt · 2022-08-19T18:19:25Z

@jpechane @smiklosovic re-tagging you in this PR.

smiklosovic · 2022-08-19T19:41:52Z

I ll do a review next week.

smiklosovic · 2022-08-23T08:48:00Z

...andra-4/src/main/java/io/debezium/connector/cassandra/AbstractCassandra4CommitLogParser.java

+    protected Cassandra4CommitLogProcessor.LogicalCommitLog commitLog;
+    protected int pollingInterval;
+
+    public AbstractCassandra4CommitLogParser(Cassandra4CommitLogProcessor.LogicalCommitLog commitLog, final List<ChangeEventQueue<Event>> queues,


Could you please move LogicalCommitLog out of Cassandra4CommitLogProcessor so it is standalone class in the same package? I find it weird that parser depends on some random stuff in log processor. Same situation with ProcessingResult.

ProcessingResult might be renamed to CommitLogProcessingResult.

smiklosovic · 2022-08-23T08:48:35Z

...andra-4/src/main/java/io/debezium/connector/cassandra/AbstractCassandra4CommitLogParser.java

+    public Cassandra4CommitLogProcessor.ProcessingResult process() {
+        if (!commitLog.exists()) {
+            LOGGER.warn("Commit log " + commitLog + " does not exist!");
+            return new Cassandra4CommitLogProcessor.ProcessingResult(commitLog, Cassandra4CommitLogProcessor.ProcessingResult.Result.DOES_NOT_EXIST);


DOES_NOT_EXIST might be imported.

smiklosovic · 2022-08-23T08:52:06Z

cassandra-4/src/main/java/io/debezium/connector/cassandra/Cassandra4CommitLogBatchParser.java

+
+            while (!commitLog.completed) {
+                if (completePrematurely) {
+                    LOGGER.info("{} completed prematurely", commitLog);


might be {} processing completed prematurely or Processing of {} completed prematurely.

smiklosovic · 2022-08-23T08:53:16Z

...andra-4/src/main/java/io/debezium/connector/cassandra/Cassandra4CommitLogRealTimeParser.java

+        try {
+            parseIndexFile(commitLog);
+            while (!commitLog.completed) {
+                LOGGER.info("Polling for completeness of idx file for: {}", commitLog.toString());


toString is not necessary.

smiklosovic · 2022-08-23T08:53:43Z

...andra-4/src/main/java/io/debezium/connector/cassandra/Cassandra4CommitLogRealTimeParser.java

+
+                CommitLogPosition commitLogPosition = null;
+                if (offset == null) {
+                    LOGGER.info("Start to read the partial file : {}", commitLog.toString());


commitLog.toString is not necessary

Also, can you please remove redundant space before the colon? Like ... partial file: {}.

smiklosovic · 2022-08-23T08:53:48Z

...andra-4/src/main/java/io/debezium/connector/cassandra/Cassandra4CommitLogRealTimeParser.java

+                    commitLogPosition = new CommitLogPosition(commitLog.commitLogSegmentId, 0);
+                }
+                else if (offset < commitLog.offsetOfEndOfLastWrittenCDCMutation) {
+                    LOGGER.info("Resume to read the partial file : {}", commitLog.toString());


commitLog.toString is not necessary

plus same space issue before colon.

smiklosovic · 2022-08-23T08:54:29Z

...andra-4/src/main/java/io/debezium/connector/cassandra/Cassandra4CommitLogRealTimeParser.java

+                    commitLogPosition = new CommitLogPosition(commitLog.commitLogSegmentId, offset);
+                }
+                else {
+                    LOGGER.info("No movement in offset in IDX file: {}", commitLog.toString());


commitLog.toString is not necessary

smiklosovic · 2022-08-23T08:55:30Z

...andra-4/src/main/java/io/debezium/connector/cassandra/Cassandra4CommitLogRealTimeParser.java

+                parseIndexFile(commitLog);
+            }
+
+            LOGGER.info("IDX file is completed for: {}", commitLog.toString());


Once there is IDX other time idx. We should be consistent in it. Also, commitLog.toString here is not necessary.

smiklosovic · 2022-08-23T08:56:03Z

...andra-4/src/main/java/io/debezium/connector/cassandra/Cassandra4CommitLogRealTimeParser.java

+            return new Cassandra4CommitLogProcessor.ProcessingResult(commitLog);
+        }
+        catch (final Exception ex) {
+            LOGGER.error("Processing of {} errored out", commitLog.toString(), ex);


also, toString is redundant

smiklosovic · 2022-08-23T08:57:00Z

core/src/main/java/io/debezium/connector/cassandra/CassandraConnectorConfig.java

+    public static final Field COMMIT_LOG_REAL_TIME_PROCESSING_ENABLED = Field.create("commit.log.real.time.processing.enabled")
+            .withType(Type.BOOLEAN)
+            .withDefault(DEFAULT_COMMIT_LOG_REAL_TIME_PROCESSING_ENABLED)
+            .withDescription("Enables the near real time processing of commit logs for Cassandra 4 by reading commit log files incrementally");


should not there be some hyphens in near real time? Like near real-time.

smiklosovic · 2022-08-23T08:57:54Z

core/src/main/java/io/debezium/connector/cassandra/CassandraConnectorConfig.java

+    public static final Field COMMIT_LOG_MARKED_COMPLETE_POLL_INTERVAL_IN_MS = Field.create("commit.log.marked.complete.poll.interval.ms")
+            .withType(Type.INT)
+            .withDefault(DEFAULT_COMMIT_LOG_MARKED_COMPLETE_POLL_INTERVAL_IN_MS)
+            .withDescription("Defines the polling interval to check for Commit Log file marked complete in Cassandra 4");


I think it should be CommitLog, like these words should be together.

smiklosovic · 2022-08-23T08:58:08Z

...andra-4/src/main/java/io/debezium/connector/cassandra/Cassandra4CommitLogRealTimeParser.java

+                LOGGER.info("Polling for completeness of idx file for: {}", commitLog.toString());
+                if (completePrematurely) {
+                    LOGGER.info("{} completed prematurely", commitLog.toString());
+                    return new Cassandra4CommitLogProcessor.ProcessingResult(commitLog, Cassandra4CommitLogProcessor.ProcessingResult.Result.COMPLETED_PREMATURELY);


might be imported

smiklosovic · 2022-08-23T08:58:38Z

...andra-4/src/main/java/io/debezium/connector/cassandra/AbstractCassandra4CommitLogParser.java

+            LOGGER.debug("finished reading commit log segments {} on position {}", logicalCommitLog, position);
+        }
+        catch (Exception e) {
+            if (commitLogTransfer.getClass().getName().equals(CassandraConnectorConfig.DEFAULT_COMMIT_LOG_TRANSFER_CLASS)) {


DEFAULT_COMMIT_LOG_TRANSFER_CLASS might be imported

smiklosovic · 2022-08-23T09:02:55Z

cassandra-4/src/main/java/io/debezium/connector/cassandra/Cassandra4CommitLogProcessor.java

+            parser = new Cassandra4CommitLogBatchParser(new LogicalCommitLog(index.toFile()), queues, metrics, this.context);
+        }
+
+        Future<ProcessingResult> future = executorService.submit(() -> parser.process());


() -> parser.process() might be replaced with method reference like .submit(parser::proces).

smiklosovic · 2022-08-23T09:05:16Z

...andra-4/src/main/java/io/debezium/connector/cassandra/AbstractCassandra4CommitLogParser.java

+
+    protected void processCommitLog(Cassandra4CommitLogProcessor.LogicalCommitLog logicalCommitLog, CommitLogPosition position) {
+        try {
+            LOGGER.debug("starting to read commit log segments {} on position {}", logicalCommitLog, position);


Could you upper-case this? I am sorry I left it like that (I think it was me). Same with the next debug log below.

smiklosovic · 2022-08-23T09:17:21Z

The main thing I am missing here is a proper test case. Without this, we can not know this actually works without testing that manually. I think the solid start would be to try to create a simple test case, something like this:

public class RealtimeTest extends AbstractCommitLogProcessorTest {

    @Override
    protected CassandraConnectorContext generateTaskContext() throws Exception {
	Properties properties = TestUtils.generateDefaultConfigMap();
	properties.put(CassandraConnectorConfig.COMMIT_LOG_REAL_TIME_PROCESSING_ENABLED, true);
	return generateTaskContext(Configuration.from(properties));
    }

    @Override
    public void initialiseData() throws Exception {

    }

    @Override
    public void verifyEvents() throws Exception {

    }
}

AbstractCommitLogProcessorTest is currently written in quite opinionated way and it does not count on the fact we might initialise and verify events differently. I think we might create another abstract class which satisfies real-time testing scenario better with refactored setup method and so on. That is where you might get quite creative, I have not explored that more deeply.

My idea of a test case is something like this:

insert few rows and wait until they are processed (like 10s or what is there, we might also change that config value for the sake of tests being faster / scanning occuring more often).
verify that rows were read
insert yet few more rows
verify only these added rows were processed

I think that after some refactoring this should be quite straightforward to do.

nitinitt · 2022-08-29T05:32:40Z

The main thing I am missing here is a proper test case. Without this, we can not know this actually works without testing that manually. I think the solid start would be to try to create a simple test case, something like this:
public class RealtimeTest extends AbstractCommitLogProcessorTest {

    @Override
    protected CassandraConnectorContext generateTaskContext() throws Exception {
	Properties properties = TestUtils.generateDefaultConfigMap();
	properties.put(CassandraConnectorConfig.COMMIT_LOG_REAL_TIME_PROCESSING_ENABLED, true);
	return generateTaskContext(Configuration.from(properties));
    }

    @Override
    public void initialiseData() throws Exception {

    }

    @Override
    public void verifyEvents() throws Exception {

    }
}
AbstractCommitLogProcessorTest is currently written in quite opinionated way and it does not count on the fact we might initialise and verify events differently. I think we might create another abstract class which satisfies real-time testing scenario better with refactored setup method and so on. That is where you might get quite creative, I have not explored that more deeply.

My idea of a test case is something like this:

insert few rows and wait until they are processed (like 10s or what is there, we might also change that config value for the sake of tests being faster / scanning occuring more often).

verify that rows were read

insert yet few more rows

verify only these added rows were processed

I think that after some refactoring this should be quite straightforward to do.

Hey @smiklosovic Thanks for the pointers on creating tests, and I totally agree to have tests to have confidence in this new feature. I was able to add a new test for real-time parsing, exactly as you mentioned in the comment above. It was little trickier than I thought particularly due to presence of 2 Cassandra YAML file(one for embedded cassandra/docker and other for running Cassandra Debezium tests). The existing tests provide coverage for CommitLogReadHandlerImpl class which is great, and this approach in new test class: CommitLogRealTimeParserTest can be leveraged to test the Cassandra4CommitLogProcessor class.
Also, have addressed all the comments as well. Hopefully the PR is in a mergeable state now.

smiklosovic · 2022-08-29T14:36:09Z

cassandra-4/src/test/java/io/debezium/connector/cassandra/CommitLogRealTimeParserTest.java

+        for (int i = 0; i < expectedEventsCount; i++) {
+            Record record = (Record) events.get(i);
+            Assert.assertEquals("Operation type must be insert", Record.Operation.INSERT, record.getOp());
+            Assert.assertEquals("Inserted key should be " + i, record.getRowData().getPrimary().get(0).value, i + keyInc);


should not this be "Inserted key should be" + (i + keyInc) ?

smiklosovic · 2022-08-29T14:37:17Z

cassandra-4/src/test/java/io/debezium/connector/cassandra/CommitLogRealTimeParserTest.java

+
+        for (int i = 0; i < expectedEventsCount; i++) {
+            Record record = (Record) events.get(i);
+            Assert.assertEquals("Operation type must be insert", Record.Operation.INSERT, record.getOp());


would be nice to actually log what that operation was if it was not insert, it is nice we know it is not insert, so what is it?

@nitinitt

this is enough but I was expecting it to be directly in that assert. Like

Record.Operation op = record.getOp(); Assert.assertEquals("Operation type must be insert but it was " + op, Record.Operation.INSERT, op);

I just do not to see the value in log and then in assert which does not log it when it fails. You would need to look into the logs first to find it there what it was.

This sounds better, have updated the assert statement. Thanks!

smiklosovic · 2022-08-29T14:42:26Z

cassandra-4/src/test/java/io/debezium/connector/cassandra/CommitLogRealTimeParserTest.java

+        String cdcLoc = DatabaseDescriptor.getCDCLogLocation();
+        LOGGER.info("CDC Location: {}", cdcLoc);
+
+        File[] commitLogs = CommitLogUtil.getIndexes(new File(cdcLoc));


what you get are not commitLogs but indexes. Hence in the next for loop, you are supposed to iterate over indexes and you are submitting an index as well. From index, the actual commit log file name is parsed.

Yes, you are right. Have renamed the variable to commitLogIndexes for better readability

smiklosovic · 2022-08-29T14:44:00Z

cassandra-4/src/test/java/io/debezium/connector/cassandra/CommitLogRealTimeParserTest.java

+        LOGGER.info("CDC Location: {}", cdcLoc);
+
+        File[] commitLogs = CommitLogUtil.getIndexes(new File(cdcLoc));
+        Thread.sleep(2000);


what is this sleep good for? Would you mind to elaborate a bit? Maybe it would be better to use Awaitility and wait for some condition to happen? Having explicit sleep times like this tend to be flaky from what I observed in the past.

Glad you asked. Firstly, this sleep is at incorrect place. It should be above L89, i.e. before reading the idx files, and you are right this sleep can cause flakiness in tests and I have replaced it with Awaitility. Cassandra periodically(as defined in cassandra.yaml file, param: commitlog_sync_period_in_ms) flushes idx files and Commit Log files to cdc location. The sleep was to make sure idx file got generated/updated, and now it will be replaced by a condition.

smiklosovic · 2022-08-29T14:45:11Z

cassandra-4/src/test/resources/docker/cassandra.yaml

@@ -222,7 +222,7 @@ cdc_enabled: true
 # separate spindle than the data directories. If not set, the default directory is
 # $CASSANDRA_HOME/data/cdc_raw.
 # cdc_raw_directory: /var/lib/cassandra/cdc_raw
-cdc_raw_directory: /var/lib/cassandra/cdc_raw_directory
+cdc_raw_directory: /var/lib/cassandra/cdc_raw


just double checking, is this what we want? (in case it was a typo)

Yes, this was intentional. It is to make sure the directory name is same as provided in cassandra-unit-for-context.yaml, so that test code can programmatically pick it up via: DatabaseDescriptor.getCDCLogLocation() method.

jpechane · 2022-08-30T03:50:22Z

cassandra-4/src/test/java/io/debezium/connector/cassandra/CommitLogRealTimeParserTest.java

+        List<Event> events = queue.poll();
+        int count = 0;
+        int maxRetryCount = 3;
+        while (events.size() != expectedEventsCount && count < maxRetryCount) {


Please use Awaitility library for this type of code like we do in other connectors.

Thanks, updated the test to use Awaitility.

…ntally, enabling real-time processing

nitinitt · 2022-08-30T20:55:43Z

Hey @jpechane, I have incorporated all the code review feedback. Please let me know if its in a mergeable/acceptable state.

jpechane · 2022-09-01T07:20:47Z

@nitinitt Applied, thanks a lot! Could you please provide a docs PR to the main repo reflecting the change in the connector behaviour? Thanks

nitinitt · 2022-09-01T17:10:40Z

@nitinitt Applied, thanks a lot! Could you please provide a docs PR to the main repo reflecting the change in the connector behaviour? Thanks

Thanks @jpechane for merging the PR. I will update the docs and create a new PR in Debezium repo. Would this feature be a part of 2.0.0.beta2 release? Would you want this PR to be applied to 1.9 series as well?

jpechane · 2022-09-01T17:13:16Z

@nitinitt Yes, it will be in 2.0.0.Beta2. There is no need for 1.9 as this is a new feature and we usually limit backporting to bugfixes only.

smiklosovic reviewed Aug 23, 2022

View reviewed changes

smiklosovic mentioned this pull request Aug 23, 2022

DBZ-5410: Ability to read commit log files incrementally in Cassandra4 #62

Closed

smiklosovic reviewed Aug 29, 2022

View reviewed changes

jpechane reviewed Aug 30, 2022

View reviewed changes

DBZ-5410 Added ability to process Cassandra 4 CommitLog files increme…

6cd35c2

…ntally, enabling real-time processing

nitinitt force-pushed the DBZ-5410-real-time-processor branch from daba56d to 6cd35c2 Compare August 30, 2022 20:37

jpechane merged commit af7c492 into debezium:main Sep 1, 2022

smiklosovic mentioned this pull request Mar 1, 2023

DBZ-2514 #43

Closed

DBZ-5410 Added ability to process Cassandra 4 Commit log files incrementally #69

DBZ-5410 Added ability to process Cassandra 4 Commit log files incrementally #69

Conversation

nitinitt commented Aug 19, 2022

nitinitt commented Aug 19, 2022

smiklosovic commented Aug 19, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

smiklosovic commented Aug 23, 2022 • edited

nitinitt commented Aug 29, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

smiklosovic Aug 30, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

smiklosovic Aug 29, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nitinitt Aug 29, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nitinitt Aug 30, 2022 • edited

Choose a reason for hiding this comment

nitinitt commented Aug 30, 2022

jpechane commented Sep 1, 2022

nitinitt commented Sep 1, 2022

jpechane commented Sep 1, 2022

smiklosovic commented Aug 23, 2022 •

edited

nitinitt commented Aug 29, 2022 •

edited

smiklosovic Aug 30, 2022 •

edited

smiklosovic Aug 29, 2022 •

edited

nitinitt Aug 29, 2022 •

edited

nitinitt Aug 30, 2022 •

edited