DBZ-6984 Add multi-task support for MongoDB connectors by adityanair-stripe · Pull Request #5925 · debezium/debezium

adityanair-stripe · 2024-10-07T20:37:58Z

https://issues.redhat.com/browse/DBZ-6984

jpechane · 2024-10-08T06:34:43Z

debezium-core/src/main/java/io/debezium/config/CommonConnectorConfig.java

        this.signalEnabledChannels = getSignalEnabledChannels(config);
        this.skippedOperations = determineSkippedOperations(config);
        this.taskId = config.getString(TASK_ID);
+        this.maxTasks = config.getInteger(TASKS_MAX, 1);


Is this really guranteed that Kafka Connect will always propagate the connector infra level config to the connector configuration?

I don't think there's any documented guarantee. It is the actual behavior though: https://github.com/apache/kafka/blob/trunk/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java#L2106

jpechane · 2024-10-08T06:37:36Z

...um-connector-mongodb/src/main/java/io/debezium/connector/mongodb/MongoDbConnectorConfig.java

                    "for data change, schema change, transaction, heartbeat event etc.")
            .withDefault(DefaultTopicNamingStrategy.class.getName());

+    public static final Field MONGODB_MULTI_TASK_ENABLED = Field.create("mongodb.multi.task.enabled")


Do we really need to have such a config option? IMHO just the setting of tasks.max > 1 should trigger the behaviour.
But as a safety concern (given we are going to merge it to stable branch) I'd keep this yet use Field.createInternal. It would be false by default in 3.0 and true in 3.1. The option could be then removed completely.

I would definitively keep an option that specifically enables THIS TYPE of multitask As in the future we might want to do multi tasking based on e.g. databases.

Agree with @jcechace on this :) There is also an existing replset based multi-tasking for Mongo, so we wanted to be very explicit about having a flag to turn the feature on/off.

@ning-stripe RS-based one was removed as it was highly problematic

jpechane · 2024-10-08T06:38:24Z

...um-connector-mongodb/src/main/java/io/debezium/connector/mongodb/MongoDbConnectorConfig.java

            .withDefault(DefaultTopicNamingStrategy.class.getName());

+    public static final Field MONGODB_MULTI_TASK_ENABLED = Field.create("mongodb.multi.task.enabled")
+            .withDisplayName("Enable Mongo DB multi-task")


Here and anywhere else, it is MongoDB not Mongo DB

jpechane · 2024-10-08T06:49:22Z

...um-connector-mongodb/src/main/java/io/debezium/connector/mongodb/MongoDbConnectorConfig.java

+            .withValidation(Field::isInteger)
+            .withDescription("Define multi-task generation to be enabled on MongoDB instance");
+
+    public static final Field MONGODB_MULTI_TASK_PREV_TASKS = Field.create("mongodb.multi.prev.tasks.max")


Are the prev and even gen options really necessary? Can't we store the number of tasksin the offsets and detect the change based on comparison of value from offests with the one configured in connector config?

prev_gen is here because we might not want to start at generation - 1. Suppose there was some problem with the current generation, we might want to start a new generation starting where the previous generation left off. Ex: generation 0 is fine, generation 1 is problematic, so we start generation 2 from generation 0's offsets. For this to work gen and prev_gen need to be explicitly defined.

Moving gen into the offset creates an issue in the above scenario because if we don't include the generation in the partition then we won't be able to get an older generation's offsets.

jpechane · 2024-10-08T06:52:08Z

debezium-connector-mongodb/src/main/java/io/debezium/connector/mongodb/MongoDbPartition.java

+        if (this.multiTaskEnabled) {
+            partition.put(TASK_ID_KEY, String.valueOf(this.taskId));
+            partition.put(MULTI_TASK_GEN_KEY, String.valueOf(this.multiTaskGen));
+            partition.put(TASK_COUNT_KEY, String.valueOf(taskCount));


Why do we need to use taks count?

We included task count in the partition to guard against accidental misconfiguration. Ex: if generation 0 has 3 tasks, and generation 1 is configured with prev_tasks=1 the connector will fail to find any offsets and exit instead of silently using the incorrect configuration.

jcechace · 2024-10-08T20:46:18Z

I'm going over this. However before this gets merged I would very much appreciate at least some basic docs describing how the system works.

ning-stripe · 2024-10-08T22:06:03Z

...nnector-mongodb/src/test/java/io/debezium/connector/mongodb/MongoDbMultiTaskConnectorIT.java

+
+import io.debezium.data.Envelope;
+
+public class MongoDbMultiTaskConnectorIT extends AbstractMongoConnectorIT {


Amazing work! The tests look so much more thorough and less hacky :D

jcechace · 2024-10-09T09:51:47Z

@adityanair-stripe I'm slowly crunching through the PR. Would it be possible to schedule a call for the next week to discuss the concept? I know we touched on it sometime last year but a lot of things changed since then -- for one the buffering cursor did not exist.

The way I see this being helpful is only when there is a significant write into a single collection and Debezium is not able to keep up in single task.

If the writes were spread across databases / collections then multitasking based on databases/collections seems to make more sense. If the load is really on single collection, wouldn't it then make more sense to introduce paralel event processing into a single task?

I'm trying to understand where exactly the bottleneck was and whether this is addressing it at the right spot -- as although this is likely a feasible solution, it also adds a lot of complexity to the code.

jcechace · 2024-10-08T20:51:45Z

...um-connector-mongodb/src/main/java/io/debezium/connector/mongodb/MultiTaskOffsetHandler.java

+ *
+ * @author Anthony Noakes
+ */
+public class MultiTaskOffsetHandler {


The methods here definitively need javadocs.

I would probably consider a different name. E.g. "TimeIntervalHandler" as these don't really seem to be offsets (in context of Kafka).

sure, I'll call it MultiTaskWindowHandler

jcechace · 2024-10-08T20:57:53Z

...-mongodb/src/main/java/io/debezium/connector/mongodb/events/BufferingChangeStreamCursor.java

+            do {
+                try (MongoChangeStreamCursor<ChangeStreamDocument<TResult>> cursor = stream.cursor()) {
+                    cursorRef.compareAndSet(null, cursor);
+                    running.set(true);
+                    noMessageIterations = 0;
+                    fetchEvents(cursor);
+                    stream = streamManager.updateStream(stream);
+                }
+                catch (InterruptedException e) {
+                    LOGGER.error("Fetcher thread interrupted", e);
+                    Thread.currentThread().interrupt();
+                    throw new DebeziumException("Fetcher thread interrupted", e);
+                }
+                catch (Throwable e) {
+                    error.set(e);
+                    LOGGER.error("Fetcher thread has failed", e);
+                    close();
+                }
+                finally {
+                    cursorRef.set(null);
+                }
+            } while (isRunning());


Why is this needed? Having a loop inside loop (since the fetchEvent call already does this) doesn't seem right. I also wonder whether it can pose a synchronisation issue and accidentally restart the streaming.

The outer loop is to move the multi-task window whenever we finish streaming all records from the current window. Within fetchEvents we call shouldUpdateStream on every new event and break out of the inner loop when it returns true (when we get an event outside the current window). Then we move the window, clean up the cursor, update the stream, and open a new cursor before starting to iterate over events again.

jcechace · 2024-10-08T20:58:18Z

...-mongodb/src/main/java/io/debezium/connector/mongodb/events/BufferingChangeStreamCursor.java

+                    cursorRef.set(null);
+                }
+            } while (isRunning());
+            close();


this definitively shouldn't be here.

jcechace · 2024-10-08T21:07:03Z

...r-mongodb/src/main/java/io/debezium/connector/mongodb/MongoDbStreamingChangeEventSource.java

+        try (var cursor = multiTaskEnabled ? BufferingChangeStreamCursor.fromIterable(stream, multiTaskOffsetHandler, taskContext, streamingMetrics, clock).start()
+                : BufferingChangeStreamCursor.fromIterable(stream, taskContext, streamingMetrics, clock).start()) {


void ternary operator in try with resources please. It's really difficult to read. Passing null argument and redirecting to overload is perfectly fine

jcechace · 2024-10-09T09:39:09Z

...um-connector-mongodb/src/main/java/io/debezium/connector/mongodb/MongoDbConnectorConfig.java

        return Module.name();
    }

+    public boolean getMultiTaskEnabled() {


isMultiTaskEnabled

github-actions · 2024-10-15T22:27:12Z

Hi @adityanair-stripe, thanks for your contribution. Please prefix the commit message(s) with the DBZ-xxx JIRA issue key.

adityanair-stripe · 2024-10-15T22:27:24Z

@jcechace

However before this gets merged I would very much appreciate at least some basic docs describing how the system works.

I've collected some of our docs and combined them into this doc. Let me know if there are any other details about the design that you'd like me to elaborate on.

Would it be possible to schedule a call for the next week to discuss the concept

Sent an email, let me know what time works for you.

The way I see this being helpful is only when there is a significant write into a single collection

Yeah, this is a common occurrence for us - one of the goals of this feature is to solve the 'hot collection' issue described here. Distributing events to tasks based on db/collection doesn't really address the issue.

github-actions · 2024-10-15T22:34:53Z

Hi @adityanair-stripe, thanks for your contribution. Please prefix the commit message(s) with the DBZ-xxx JIRA issue key.

jcechace · 2024-10-31T14:20:38Z

this doc

This seems not to be accessible. Can you either make it public or share with me directly?

github-actions · 2024-11-08T19:09:40Z

Hi @adityanair-stripe, thanks for your contribution. Please prefix the commit message(s) with the DBZ-xxx JIRA issue key.

github-actions · 2024-11-08T19:21:45Z

Hi @adityanair-stripe, thanks for your contribution. Please prefix the commit message(s) with the DBZ-xxx JIRA issue key.

github-actions · 2024-11-08T23:23:27Z

Hi @adityanair-stripe, thanks for your contribution. Please prefix the commit message(s) with the DBZ-xxx JIRA issue key.

Naros · 2026-01-06T13:59:01Z

❌ Developer Certificate of Origin (DCO) check failed.

Hi @adityanair-stripe, please sign off all commits with:

git commit -s

If pull request commits are not signed off, the pull request cannot be merged. For more information about why this is required, please see our blog about contribution requirement changes.

Signed-off-by: Aditya Nair <[email protected]>

jpechane requested review from ani-sha and jcechace October 8, 2024 06:31

jpechane reviewed Oct 8, 2024

View reviewed changes

ning-stripe reviewed Oct 8, 2024

View reviewed changes

jcechace reviewed Oct 10, 2024

View reviewed changes

adityanair-stripe requested a review from jcechace November 9, 2024 00:39

adityanair-stripe force-pushed the DBZ-6984 branch from 801dc64 to b38753b Compare January 23, 2025 20:19

DBZ-6984 Add multi-task support for MongoDB connectors

d0cea5f

Signed-off-by: Aditya Nair <[email protected]>

adityanair-stripe force-pushed the DBZ-6984 branch from b38753b to 24d07db Compare March 6, 2026 17:46

DBZ-6984 Address review feedback

e6a4bf4

Signed-off-by: Aditya Nair <[email protected]>

adityanair-stripe force-pushed the DBZ-6984 branch from 24d07db to e6a4bf4 Compare March 6, 2026 19:06

Naros requested review from jpechane and removed request for ani-sha March 10, 2026 17:23


		import io.debezium.data.Envelope;

		public class MongoDbMultiTaskConnectorIT extends AbstractMongoConnectorIT {

		try (var cursor = multiTaskEnabled ? BufferingChangeStreamCursor.fromIterable(stream, multiTaskOffsetHandler, taskContext, streamingMetrics, clock).start()
		: BufferingChangeStreamCursor.fromIterable(stream, taskContext, streamingMetrics, clock).start()) {

Uh oh!

Conversation

adityanair-stripe commented Oct 7, 2024

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

adityanair-stripe Oct 8, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jcechace commented Oct 8, 2024

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jcechace commented Oct 9, 2024

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

github-actions bot commented Oct 15, 2024

Uh oh!

adityanair-stripe commented Oct 15, 2024

Uh oh!

github-actions bot commented Oct 15, 2024

Uh oh!

jcechace commented Oct 31, 2024

Uh oh!

github-actions bot commented Nov 8, 2024

Uh oh!

github-actions bot commented Nov 8, 2024

Uh oh!

github-actions bot commented Nov 8, 2024

Uh oh!

Naros commented Jan 6, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

adityanair-stripe Oct 8, 2024 •

edited

Loading