Fix for kafka source not committing offsets issue #3231 #3232

hshardeesi · 2023-08-24T04:32:02Z

Description

This PR fixes following issues:

Commit offsets when a kafka partition's beginning offset is non-zero.
Fix numRecordsCommitted metric.
Fix shutdown sequence for all topics+workers.

Following tests were conducted with these changes:

Ingest 10M records from a kafka topic + group membership mutations.
Ingest 12 M records from 2 kafka topics in single pipeline.
Ingest 10M records from 2 different topics in 2 different pipelines.
2 pipelines part of same consumer group ingesting 10 M records.
Shutdown pipeline in the middle of ingestion and resume and make sure all data is ingested.
Ingest data without acknowledgments enabled.
Ingest data with auto_offset_reset=earliest/latest.

Issues Resolved

Resolves #3231

Check List

New functionality includes testing.
New functionality has a documentation issue. Please link to it in this PR.
- New functionality has javadoc added
Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

engechas · 2023-08-24T14:35:28Z

...c/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaSourceCustomConsumer.java

-            if (partitionEpoch == 0) {
-                LOG.info("Skipping partition {}, lost ownership", topicPartition);
+            if (acknowledgementsEnabled && partitionEpoch == 0) {
+                //ToDo: Add metric


Would the metric here be useful for debugging? I'm wondering if it's worth just adding now

I didn't have time to add a new metric. I'll see if I can add now, but there are other metrics and logs from which it can be inferred if we are hitting this condition. Ideally, it should not happen and an assert is more appropriate here.

engechas · 2023-08-24T14:37:30Z

...c/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaSourceCustomConsumer.java

+            final Map<TopicPartition, Long> endOffsets = consumer.endOffsets(partitions);
+            for (TopicPartition topicPartition : partitions) {
+                final OffsetAndMetadata offsetAndMetadata = committedOffsets.get(topicPartition);
+                LOG.info("Assigned partition {} with offsets: beginningOffset: {}, endOffset: {}, committedOffset: {}",


Minor: this might be better off as a debug log rather than info

this is very crucial log for debugging, I wish we had it from the beginning. If debug shows up by default then I can change it to debug level else leave it as INFO.

engechas · 2023-08-24T14:38:56Z

...kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/source/KafkaSource.java

-            stopConsumer(topicExecutorService.get(topic), topicConsumer.get(topic));
-        });
-        LOG.info("Consumer shutdown successfully...");
+        final long shutdownWaitTime = calculateLongestThreadWaitingTime();


What's the upper limit on this?

Looking at calculateLongestThreadWaitingTime() implementation it looks like it can be atmost 5 seconds (default value) unless user has configured to be higher number.

dlvenable

Thanks for the contribution @hshardeesi !

There is a lot here that could be unit tested to ensure that the behavior remains correct.

dlvenable · 2023-08-24T14:47:31Z

...main/java/org/opensearch/dataprepper/plugins/kafka/consumer/TopicPartitionCommitTracker.java

@@ -68,6 +90,7 @@ public OffsetAndMetadata addCompletedOffsets(final Range<Long> offsetRange) {
            Long maxValue = offsetMinMap.get(committedOffset).getMaximum();
            if (maxValue != committedOffset) {
                offsetMinMap.remove(committedOffset);
+                committedRecordCount += (maxValue - committedOffset);


These are fairly critical calculations. Can you add unit tests for these with some of the difference scenarios? e.g. start at 0, start at non-zero.

Sure I'll add unit test for this and other scenarios you pointed out below.

dlvenable · 2023-08-24T14:48:58Z

...c/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaSourceCustomConsumer.java

+                LOG.info("Assigned partition {} with offsets: beginningOffset: {}, endOffset: {}, committedOffset: {}",
+                        topicPartition, getTopicPartitionOffset(beginningOffsets, topicPartition),
+                        getTopicPartitionOffset(endOffsets, topicPartition),
+                        Objects.isNull(offsetAndMetadata) ? "-" : offsetAndMetadata.offset());


This line isn't assigning anything. I think you can remove it.

dlvenable · 2023-08-24T14:51:01Z

...c/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaSourceCustomConsumer.java

@@ -238,6 +232,21 @@ void processAcknowledgedOffsets() {
        acknowledgedOffsets.clear();
    }

+    private void updateCommitCount(final TopicPartition topicPartition, final OffsetAndMetadata offsetAndMetadata) {


It appears this only updates the metric, but the name implies it updates an actual count. Perhaps rename to updateCommitCountMetric.

dlvenable · 2023-08-24T14:51:32Z

...c/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaSourceCustomConsumer.java

+                                    partitionCommitTrackerMap.get(partitionId).addCompletedOffsets(offsetRange.getOffsets());
+                            updateOffsetsToCommit(partition, offsetAndMetadata);
+                        } else {
+                            LOG.error("!! Commit tracker not found for topic: {} partition: {}", partition.topic(), partitionId);


I don't think we need these exclamation marks. The log level is ERROR which is sufficient here.

dlvenable · 2023-08-24T14:52:16Z

...c/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaSourceCustomConsumer.java

-                            updateOffsetsToCommit(partition,
-                                    new OffsetAndMetadata(offsetRange.getOffsets().getMaximum() + 1),
-                                    offsetRange.getOffsets()));
+                    offsets.forEach((partition, offsetRange) -> {


We should be unit testing this behavior.

hshardeesi · 2023-08-24T16:37:23Z

Thanks for the contribution @hshardeesi !

There is a lot here that could be unit tested to ensure that the behavior remains correct.

…3231 Signed-off-by: Hardeep Singh <[email protected]>

dlvenable

Thanks for updating the tests!

Signed-off-by: Hardeep Singh <[email protected]> (cherry picked from commit 89e8f39)

Signed-off-by: Hardeep Singh <[email protected]> (cherry picked from commit 89e8f39) Co-authored-by: Hardeep Singh <[email protected]>

hshardeesi requested review from chenqi0805, engechas, graytaylor0, dinujoh, kkondaka, asifsmohammed, dlvenable and oeyh as code owners August 24, 2023 04:32

engechas reviewed Aug 24, 2023

View reviewed changes

dlvenable requested changes Aug 24, 2023

View reviewed changes

hshardeesi closed this Aug 24, 2023

hshardeesi reopened this Aug 24, 2023

Fix for kafka source not committing offsets issue opensearch-project#…

6129499

…3231 Signed-off-by: Hardeep Singh <[email protected]>

hshardeesi force-pushed the kafka-commitoffsets branch from 79072b5 to 6129499 Compare August 24, 2023 18:14

hshardeesi requested review from dlvenable and engechas August 24, 2023 18:20

engechas approved these changes Aug 24, 2023

View reviewed changes

dlvenable approved these changes Aug 24, 2023

View reviewed changes

dlvenable added the backport 2.4 label Aug 24, 2023

dlvenable merged commit 89e8f39 into opensearch-project:main Aug 24, 2023
25 of 26 checks passed

opensearch-trigger-bot bot mentioned this pull request Aug 24, 2023

[Backport 2.4] Fix for kafka source not committing offsets issue #3231 #3239

Merged

opensearch-trigger-bot bot pushed a commit that referenced this pull request Aug 24, 2023

Fix for kafka source not committing offsets issue #3231 (#3232)

84e9473

Signed-off-by: Hardeep Singh <[email protected]> (cherry picked from commit 89e8f39)

dlvenable pushed a commit that referenced this pull request Aug 24, 2023

Fix for kafka source not committing offsets issue #3231 (#3232) (#3239)

6413a3d

Signed-off-by: Hardeep Singh <[email protected]> (cherry picked from commit 89e8f39) Co-authored-by: Hardeep Singh <[email protected]>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fix for kafka source not committing offsets issue #3231 #3232

Fix for kafka source not committing offsets issue #3231 #3232

hshardeesi commented Aug 24, 2023

engechas Aug 24, 2023

hshardeesi Aug 24, 2023

engechas Aug 24, 2023

hshardeesi Aug 24, 2023

engechas Aug 24, 2023

hshardeesi Aug 24, 2023

dlvenable left a comment

dlvenable Aug 24, 2023

hshardeesi Aug 24, 2023

dlvenable Aug 24, 2023

dlvenable Aug 24, 2023

dlvenable Aug 24, 2023

dlvenable Aug 24, 2023

hshardeesi commented Aug 24, 2023

dlvenable left a comment

Fix for kafka source not committing offsets issue #3231 #3232

Fix for kafka source not committing offsets issue #3231 #3232

Conversation

hshardeesi commented Aug 24, 2023

Description

Issues Resolved

Check List

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

dlvenable left a 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

hshardeesi commented Aug 24, 2023

dlvenable left a comment

Choose a reason for hiding this comment