[Improve][Kafka] kafka source refactored some reader read logic #6408

Carl-Zhou-CN · 2024-02-28T09:13:36Z

An attempt to improve kafka source with reference to flink kafka source

Purpose of this pull request

Does this PR introduce any user-facing change?

How was this patch tested?

Check list

If any new Jar binary package adding in your PR, please add License Notice according
New License Guide
If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
If you are contributing the connector code, please check that the following files are updated:
1. Update change log that in connector document. For more details you can refer to connector-v2
2. Update plugin-mapping.properties and add new connector information in it
3. Update the pom file of seatunnel-dist
Update the release-note.

Carl-Zhou-CN · 2024-05-29T10:29:16Z

@zhilinli123 @hailin0 @Hisoka-X @EricJoy2048 @TyrantLucifer cc

hailin0 · 2024-06-15T14:32:03Z

Please fix the conflict

Carl-Zhou-CN · 2024-08-08T10:12:13Z

Please fix the conflict

done

Carl-Zhou-CN · 2024-08-08T10:12:36Z

@hailin0

Hisoka-X · 2024-08-24T02:47:12Z

...nnel-common/src/main/java/org/apache/seatunnel/common/utils/TemporaryClassLoaderContext.java

+    // ------------------------------------------------------------------------
+


Suggested change

// ------------------------------------------------------------------------

Hisoka-X · 2024-08-24T02:48:29Z

...nnel-common/src/main/java/org/apache/seatunnel/common/utils/TemporaryClassLoaderContext.java

+ */
+package org.apache.seatunnel.common.utils;
+
+public final class TemporaryClassLoaderContext implements AutoCloseable {


I like this tool!

Hisoka-X · 2024-08-24T02:57:50Z

...ava/org/apache/seatunnel/connectors/seatunnel/common/source/reader/fetcher/SplitFetcher.java

+    }
+
+    private void enqueueTaskUnsafe(SplitFetcherTask task) {
+        assert lock.isHeldByCurrentThread();


As default, assert not working. https://docs.oracle.com/cd/E19683-01/806-7930/6jgp65ikq/index.html

I believe that this ReentrantLock usage is the same as depicted in the diagram, only occurring during testing and debugging. It can be disabled in production, but of course, it can also be enabled for added safety.

cc @EricJoy2048 @CheneyYin

Hisoka-X · 2024-08-24T03:03:58Z

Please share more information about why we need refactor kafka? And what we can get after refactor?

hailin0 · 2024-08-24T03:15:04Z

Please share more information about why we need refactor kafka? And what we can get after refactor?

+1

Carl-Zhou-CN · 2024-08-24T04:09:35Z

Please share more information about why we need refactor kafka? And what we can get after refactor?

Firstly, I believe there is room for improvement in the previous approach where a new consumer thread is created for each split, and each consumer thread polls in a loop. Enhancements could involve having one consumer thread per parallelism level, ensuring clearer offset management, and promptly removing consumed partitions to reduce overhead.
Secondly, a more seamless integration of streaming and batch processing can be achieved, eliminating the need for consumer threads to differentiate between streaming and batch behaviors.

Hisoka-X · 2024-08-24T09:38:35Z

Firstly, I believe there is room for improvement in the previous approach where a new consumer thread is created for each split, and each consumer thread polls in a loop. Enhancements could involve having one consumer thread per parallelism level, ensuring clearer offset management, and promptly removing consumed partitions to reduce overhead.
Secondly, a more seamless integration of streaming and batch processing can be achieved, eliminating the need for consumer threads to differentiate between streaming and batch behaviors.

Make sense to me. Thanks @Carl-Zhou-CN !

Hisoka-X · 2024-08-27T03:01:57Z

Checking...

...-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSource.java

Hisoka-X · 2024-08-27T06:31:03Z

...fka-e2e/src/test/resources/kafka/kafkasource_group_offset_to_console_with_commit_offset.conf

              },
              {
                rule_type = MAX
-                rule_value = 149
+                rule_value = 99


Any reason for change test case config?

The original test case did not play a role because he did not read the data

Maybe there is no suitable comment. The purpose of this test case is not to read any data and ensure that the offset can still be submitted to Kafka normally. Please refer #7312
cc @hailin0

Hisoka-X · 2024-08-27T06:34:29Z

...ava/org/apache/seatunnel/connectors/seatunnel/common/source/reader/fetcher/SplitFetcher.java

@@ -234,4 +234,19 @@ private void addTaskUnsafe(SplitFetcherTask task) {
        taskQueue.add(task);
        nonEmpty.signal();
    }
+
+    public void enqueueTask(SplitFetcherTask task) {


why not direct use

seatunnel/seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/source/reader/fetcher/SplitFetcher.java

Line 108 in c6f627f

public void addTask(@NonNull SplitFetcherTask task) {

?

ok, the new method is redundant

Hisoka-X

LGTM. Thanks @Carl-Zhou-CN

hailin0 · 2024-09-03T15:31:46Z

...src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaRecordEmitter.java

+            } else {
+                deserializationSchema.deserialize(consumerRecord.value(), outputCollector);
+            }
+            splitState.setCurrentOffset(consumerRecord.offset() + 1);


Why do we need +1?

Please add a comment

hailin0

LGTM

hailin0

LGTM

Hisoka-X

LGTM if ci passes. Thanks @Carl-Zhou-CN !

Carl-Zhou-CN marked this pull request as ready for review May 13, 2024 07:40

Carl-Zhou-CN added the discuss label May 13, 2024

hailin0 added kafka improve labels May 13, 2024

Carl-Zhou-CN changed the title ~~[Improve] kafka source improve~~ [Improve][Kafka] kafka source refactored some reader read logic May 13, 2024

Carl-Zhou-CN force-pushed the kafka-test branch from cc14bf4 to 45d061a Compare May 29, 2024 02:20

Carl-Zhou-CN force-pushed the kafka-test branch from 2ae71dd to c6f8cc2 Compare August 8, 2024 10:07

github-actions bot added connectors-v2 e2e api and removed kafka labels Aug 8, 2024

Carl-Zhou-CN added 3 commits August 8, 2024 18:10

[Improve] kafka source improve

37880b7

[BugFix][Kafka] Perfect the code

3ca6a3b

[BugFix][Kafka] fix

30a76e5

Carl-Zhou-CN force-pushed the kafka-test branch from c6f8cc2 to 30a76e5 Compare August 8, 2024 10:11

Carl-Zhou-CN and others added 7 commits August 8, 2024 19:15

[Improve][Kafka] kafka source refactored some reader read logic

ad8ddf3

fix

13e5a33

fix

b3d2c41

fix

bd295f0

fix

7ca3d49

Merge branch 'apache:dev' into kafka-test

0607810

fix

a3827ad

Hisoka-X added this to the 2.3.8 milestone Aug 23, 2024

Hisoka-X reviewed Aug 24, 2024

View reviewed changes

Hisoka-X removed this from the 2.3.8 milestone Aug 24, 2024

fix

d84dabe

Hisoka-X added this to the 2.3.8 milestone Aug 26, 2024

Carl-Zhou-CN requested a review from Hisoka-X August 26, 2024 07:27

Hisoka-X reviewed Aug 27, 2024

View reviewed changes

fix

e8ce65b

github-actions bot added the kafka label Aug 28, 2024

Hisoka-X previously approved these changes Aug 30, 2024

View reviewed changes

github-actions bot added approved reviewed labels Aug 30, 2024

hailin0 reviewed Sep 3, 2024

View reviewed changes

hailin0 reviewed Sep 4, 2024

View reviewed changes

fix

65e229c

Carl-Zhou-CN dismissed Hisoka-X’s stale review via 65e229c September 4, 2024 06:08

github-actions bot removed approved reviewed labels Sep 4, 2024

hailin0 approved these changes Sep 4, 2024

View reviewed changes

github-actions bot added approved reviewed labels Sep 4, 2024

Hisoka-X approved these changes Sep 4, 2024

View reviewed changes

Merge branch 'apache:dev' into kafka-test

adec234

hailin0 merged commit 10598b6 into apache:dev Sep 5, 2024
3 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Improve][Kafka] kafka source refactored some reader read logic #6408

[Improve][Kafka] kafka source refactored some reader read logic #6408

Carl-Zhou-CN commented Feb 28, 2024

Carl-Zhou-CN commented May 29, 2024

hailin0 commented Jun 15, 2024

Carl-Zhou-CN commented Aug 8, 2024

Carl-Zhou-CN commented Aug 8, 2024

Hisoka-X Aug 24, 2024

Carl-Zhou-CN Aug 26, 2024

Hisoka-X Aug 24, 2024

Hisoka-X Aug 24, 2024

Carl-Zhou-CN Aug 26, 2024

Carl-Zhou-CN Aug 26, 2024

Hisoka-X Aug 26, 2024

Hisoka-X commented Aug 24, 2024

hailin0 commented Aug 24, 2024

Carl-Zhou-CN commented Aug 24, 2024

Hisoka-X commented Aug 24, 2024

Hisoka-X commented Aug 27, 2024

Hisoka-X Aug 27, 2024

Carl-Zhou-CN Aug 27, 2024

Hisoka-X Aug 27, 2024

Hisoka-X Aug 27, 2024

Carl-Zhou-CN Aug 27, 2024

Hisoka-X left a comment

hailin0 Sep 3, 2024

hailin0 Sep 4, 2024

hailin0 left a comment

hailin0 left a comment

Hisoka-X left a comment

		// ------------------------------------------------------------------------

[Improve][Kafka] kafka source refactored some reader read logic #6408

[Improve][Kafka] kafka source refactored some reader read logic #6408

Conversation

Carl-Zhou-CN commented Feb 28, 2024

Purpose of this pull request

Does this PR introduce any user-facing change?

How was this patch tested?

Check list

Carl-Zhou-CN commented May 29, 2024

hailin0 commented Jun 15, 2024

Carl-Zhou-CN commented Aug 8, 2024

Carl-Zhou-CN commented Aug 8, 2024

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

Hisoka-X commented Aug 24, 2024

hailin0 commented Aug 24, 2024

Carl-Zhou-CN commented Aug 24, 2024

Hisoka-X commented Aug 24, 2024

Hisoka-X commented Aug 27, 2024

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

Hisoka-X left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hailin0 left a comment

Choose a reason for hiding this comment

hailin0 left a comment

Choose a reason for hiding this comment

Hisoka-X left a comment

Choose a reason for hiding this comment