Kafka with topicPattern can ignore old offsets spuriously #16190

zachjsh · 2024-03-22T22:36:43Z

Description

This fixes an issue in which updating a kafka streaming supervisors topic from single to multi-topic (pattern), or vice versa, could cause old offsets to be ignored spuriously. The details of the bug and how it manifests is in the linked issue. To fix the issue, this change properly handles using previously found partition offsets, in the case where the old offsets were recorded and stored by a streaming supervisor with or without multi topic enabled, and the new streaming supervisor has multi-topic enabled or disabled.

This PR has:

abhishekagarwal87 · 2024-04-01T05:31:54Z

...dexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java

@@ -444,4 +447,56 @@ public KafkaSupervisorTuningConfig getTuningConfig()
  {
    return spec.getTuningConfig();
  }
+
+  @Override
+  protected Map<KafkaTopicPartition, Long> getOffsetsFromMetadataStorage()


Instead of overriding this method, you should just override the checkSourceMetadataMatch method.

or rename this method appropriately so that callers know that its also filtering the spurious stored offsets

Doesn't this fail when updateDataSourceMetadataWithHandle is called later on since that too will match the committed metadata with the new metadata?

Updated logic for the KafkaDataSourceMetadata so that it handles topicPattern during the matching check in updateDataSourceMetadataWithHandle, thanks for pointing this out.

abhishekagarwal87 · 2024-04-01T05:38:00Z

...dexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java

+              : getIoConfig().getStream().equals(matchValue);
+
+          if (!match && !topicMisMatchLogged.contains(matchValue)) {
+            log.warn(


This could happen when going from multi-topic to single-topic? Will these bad offsets get cleared automatically?

going from multi-topic to single topic, if the multi-topic sequence numbers contained any offsets for streams that do not match the single topic name, the new metadata state will have these sequence offsets removed. However this causes that matches method in the kafka metadata to return false, which will ultimately lead to failure to publish segments, just as going from single topic -> another single topic when there were sequence offsets stored for the first single topic would. I think this is the behavior we want. Users should be explicit when sequence offsets are lost due to config change, and should be forced to reset the respective counters needed, imo. Let me know what you think

Yeah, having the user reset the offsets explicitly in this scenario when there's no match at all makes sense to me.

...ice/src/main/java/org/apache/druid/indexing/kafka/KafkaSeekableStreamEndSequenceNumbers.java

...e/src/main/java/org/apache/druid/indexing/kafka/KafkaSeekableStreamStartSequenceNumbers.java

...exing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaDataSourceMetadataTest.java

* add tests

...dexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java

abhishekrb19 · 2024-04-12T15:43:41Z

...dexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java

+              ? pattern.matcher(matchValue).matches()
+              : getIoConfig().getStream().equals(matchValue);
+
+          if (!match && !topicMisMatchLogged.contains(matchValue)) {


The function seems a bit complex to follow, and there's a risk of introducing bugs due to the multiple if...else conditionals intertwined with multi/single topic logic. Could we maybe add a single block each for the multi topic and single topic logic or make a function each?

simplified, let me know if ok now.

...dexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java

abhishekrb19 · 2024-04-12T15:58:48Z

...dexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java

@@ -444,4 +447,56 @@ public KafkaSupervisorTuningConfig getTuningConfig()
  {
    return spec.getTuningConfig();
  }
+


A brief javadoc for this function would be useful and would clarify the need for this override from the base implementation and the filtering behavior

abhishekrb19 · 2024-04-12T16:06:54Z

...e/src/main/java/org/apache/druid/indexing/kafka/KafkaSeekableStreamStartSequenceNumbers.java

+    if (this.getClass() != other.getClass()) {
+      throw new IAE(
+          "Expected instance of %s, got %s",
+          this.getClass().getName(),
+          other.getClass().getName()
+      );
+    }


nit: adding a function validateSequenceNumbersBaseType() in the base class should remove this code duplication in 4-5 places

abhishekrb19 · 2024-04-12T16:11:51Z

...ice/src/main/java/org/apache/druid/indexing/kafka/KafkaSeekableStreamEndSequenceNumbers.java

+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+@JsonTypeName(SeekableStreamEndSequenceNumbers.TYPE)


Same here. How is this jackson type information used?

added information about serialization to javadoc, let me know if ok

abhishekrb19 · 2024-04-12T16:17:47Z

...dexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java

+              : getIoConfig().getStream().equals(matchValue);
+
+          if (!match && !topicMisMatchLogged.contains(matchValue)) {
+            log.warn(


Yeah, having the user reset the offsets explicitly in this scenario when there's no match at all makes sense to me.

...e/src/main/java/org/apache/druid/indexing/kafka/KafkaSeekableStreamStartSequenceNumbers.java

AmatyaAvadhanula

Overall logic and tests LGTM!

...dexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java

zachjsh added 2 commits March 22, 2024 18:02

* fix

ded6ff3

* simplify

256af17

github-actions bot added Area - Streaming Ingestion Area - Ingestion labels Mar 22, 2024

zachjsh requested a review from abhishekagarwal87 March 22, 2024 22:39

zachjsh added 3 commits March 22, 2024 18:53

* simplify tests

d003b67

* update matches function definition for Kafka Datasource Metadata

a81ffeb

* add matchesOld

222a106

abhishekagarwal87 reviewed Apr 1, 2024

View reviewed changes

* override matches and plus for kafka based metadata / sequence numbers

0dc5537

zachjsh commented Apr 3, 2024

View reviewed changes

...ice/src/main/java/org/apache/druid/indexing/kafka/KafkaSeekableStreamEndSequenceNumbers.java Show resolved Hide resolved

zachjsh commented Apr 3, 2024

View reviewed changes

...e/src/main/java/org/apache/druid/indexing/kafka/KafkaSeekableStreamStartSequenceNumbers.java Show resolved Hide resolved

github-advanced-security bot found potential problems Apr 3, 2024

View reviewed changes

...exing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaDataSourceMetadataTest.java Fixed Show fixed Hide fixed

...exing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaDataSourceMetadataTest.java Fixed Show fixed Hide fixed

zachjsh added 4 commits April 5, 2024 15:58

* implement minus

874c3b8

* add tests

* fix failing tests

ef4de3f

* remove TODO comments

90e2374

* simplfy and add comments

2e1a916

zachjsh requested a review from abhishekagarwal87 April 5, 2024 21:44

zachjsh added 3 commits April 5, 2024 17:46

* remove unused variable in tests

894de9c

* remove unneeded function

4fe12a2

Merge remote-tracking branch 'apache/master' into kafa-topicPattern-fix

b3eeb84

zachjsh requested review from AmatyaAvadhanula and abhishekrb19 April 9, 2024 15:54

abhishekrb19 reviewed Apr 12, 2024

View reviewed changes

AmatyaAvadhanula approved these changes Apr 15, 2024

View reviewed changes

zachjsh added 5 commits April 15, 2024 19:13

* add serde tests

d59c1a1

* more stuff

827e59c

* address review comments

750b754

Merge remote-tracking branch 'apache/master' into kafa-topicPattern-fix

ad2cbfc

* remove unneeded code.

a4eb4a9

github-advanced-security bot found potential problems Apr 16, 2024

View reviewed changes

...dexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java Dismissed Show dismissed Hide dismissed

zachjsh requested a review from abhishekrb19 April 16, 2024 17:52

zachjsh merged commit 2351f03 into apache:master Apr 17, 2024
85 checks passed

zachjsh deleted the kafa-topicPattern-fix branch April 17, 2024 14:00

adarshsanjeev added this to the 30.0.0 milestone May 6, 2024

adarshsanjeev mentioned this pull request May 28, 2024

[DRAFT] 30.0.0 release notes #16505

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Kafka with topicPattern can ignore old offsets spuriously #16190

Kafka with topicPattern can ignore old offsets spuriously #16190

zachjsh commented Mar 22, 2024 •

edited

Loading

abhishekagarwal87 Apr 1, 2024

abhishekagarwal87 Apr 1, 2024

abhishekagarwal87 Apr 1, 2024

zachjsh Apr 17, 2024

abhishekagarwal87 Apr 1, 2024

zachjsh Apr 9, 2024

abhishekrb19 Apr 12, 2024 •

edited

Loading

abhishekrb19 Apr 12, 2024

zachjsh Apr 16, 2024

abhishekrb19 Apr 12, 2024

zachjsh Apr 16, 2024

abhishekrb19 Apr 12, 2024

zachjsh Apr 16, 2024

abhishekrb19 Apr 12, 2024

zachjsh Apr 16, 2024

abhishekrb19 Apr 12, 2024 •

edited

Loading

AmatyaAvadhanula left a comment

Kafka with topicPattern can ignore old offsets spuriously #16190

Kafka with topicPattern can ignore old offsets spuriously #16190

Conversation

zachjsh commented Mar 22, 2024 • edited Loading

Description

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

abhishekrb19 Apr 12, 2024 • edited Loading

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

abhishekrb19 Apr 12, 2024 • edited Loading

Choose a reason for hiding this comment

AmatyaAvadhanula left a comment

Choose a reason for hiding this comment

zachjsh commented Mar 22, 2024 •

edited

Loading

abhishekrb19 Apr 12, 2024 •

edited

Loading

abhishekrb19 Apr 12, 2024 •

edited

Loading