[Fix][Connecotr-kafka] Fix kafka IllegalArgumentException when offset is -1#9376
Conversation
There was a problem hiding this comment.
Pull Request Overview
This PR fixes a Kafka IllegalArgumentException in the connector by handling partitions with negative offsets during batch mode. The key changes include adding a new configuration file for testing the partition skip behavior, updating the KafkaIT test file with new test data and test cases, and modifying the KafkaSourceSplitEnumerator to skip partitions with a start offset of -1.
Reviewed Changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 1 comment.
| File | Description |
|---|---|
| seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_timestamp_to_console_skip_partition.conf | Added a new configuration file for testing the skip partition scenario |
| seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java | Updated test data generation and added a new test for the skip partition scenario |
| seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceSplitEnumerator.java | Modified the enumerator logic to skip partitions with an offset of -1 in non-streaming mode |
| return new DefaultSeaTunnelRowSerializer( | ||
| topicExtractor(topic, rowType, format), | ||
| partitionExtractor(null), | ||
| partitionNativeExtractor(rowType), |
There was a problem hiding this comment.
I find this method only used by test case. Could you move it into KafkaIT?
There was a problem hiding this comment.
the all method in DefaultSeaTunnelRowSerializer,only remove new DefaultSeaTunnelRowSerializer () not work
topicExtractor(topic, rowType, format),
partitionNativeExtractor(rowType),
timestampExtractor(rowType),
keyExtractor(null, rowType, format, null, null),
valueExtractor(rowType, format, delimiter, pluginConfig),
headersExtractor()There was a problem hiding this comment.
Ok, let's change the method name create to createWithPartitionAndTimestampFields
|
cc @Carl-Zhou-CN as well. |
Hisoka-X
left a comment
There was a problem hiding this comment.
LGTM if ci passes. Thanks @WenDing-Y
| if (pendingSplit.containsKey(key)) { | ||
| pendingSplit.get(key).setStartOffset(value); | ||
| } | ||
| if (!isStreamingMode && value < 0) { |
There was a problem hiding this comment.
Will it only happen in batch jobs
There was a problem hiding this comment.
Yes, this logic only applies to batch processing scenarios. In the case of stream processing, since tasks run continuously and the source data may also change over time, skipping certain partitions carelessly could lead to unexpected or strange issues.
There was a problem hiding this comment.
Could you help add a corresponding E2E IT for the streaming task as well
There was a problem hiding this comment.
What should be the focus of the test? When the offset is -1, can all partitions be consumed normally without any partition being skipped?
There was a problem hiding this comment.
Sorry, I see that the existing tests have already covered this scenario,Thanks @WenDing-Y
Purpose of this pull request
fix #9375
Does this PR introduce any user-facing change?
How was this patch tested?
Check list
New License Guide