KAFKA-18013: Add AutoOffsetResetStrategy internal class #17858

omkreddy · 2024-11-19T12:14:19Z

This is preparatory change for KIP-1106.

This PR:

Deprecates OffsetResetStrategy enum
Adds new internal class AutoOffsetResetStrategy
Replaces all OffsetResetStrategy enum usages with AutoOffsetResetStrategy
Deprecate old/Add new constructors to MockConsumer

There are no functionality changes. We will add duration based offset reset support in subsequent PRs.

mjsax

Made a pass

mjsax · 2024-11-20T20:28:13Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java


            // This may be null if the task we are currently processing was apart of a named topology that was just removed.
            // TODO KAFKA-13713: keep the StreamThreads and TopologyMetadata view of named topologies in sync until final thread has acked
            if (offsetResetStrategy != null) {
-                switch (offsetResetStrategy) {
-                    case EARLIEST:
+                switch (offsetResetStrategy.name()) {


Why do we need to switch to name() here and do a string comparison? Can't we keep an enum comparison?

As mentioned in other comment, enum wont help much here. We will be having different instances DurationBasedStrategy class. I have the updated code to avoid string comparisons.

mjsax · 2024-11-20T20:29:31Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AutoOffsetResetStrategy.java

+public class AutoOffsetResetStrategy {
+    public static final String EARLIEST_STRATEGY_NAME = "earliest";
+    public static final String LATEST_STRATEGY_NAME = "latest";
+    public static final String NONE_STRATEGY_NAME = "none";


Should we still have an internal enum for the different strategies, to avoid doing string comparisons?

Enum wont help us as we still need to parse strings like `by_duration:PnDTnHnMn. n>. I prefer to keep string names for now. I will update in next PR if required. I will update code avoid string comparison in SteamsThread class

The idea would be to have an enum plus an Optional<Duration> which is only used when the enum value is "by-duration".

mjsax · 2024-11-20T20:32:58Z

...s/src/test/java/org/apache/kafka/clients/consumer/internals/AutoOffsetResetStrategyTest.java

+        assertTrue(AutoOffsetResetStrategy.isValid("none"));
+        assertFalse(AutoOffsetResetStrategy.isValid("invalid"));
+        assertFalse(AutoOffsetResetStrategy.isValid("LATEST"));
+        assertFalse(AutoOffsetResetStrategy.isValid(""));


Should we add a test case for null ?

mjsax · 2024-11-20T20:33:23Z

...s/src/test/java/org/apache/kafka/clients/consumer/internals/AutoOffsetResetStrategyTest.java

+        assertEquals(AutoOffsetResetStrategy.NONE, AutoOffsetResetStrategy.valueOf("none"));
+        assertThrows(IllegalArgumentException.class, () -> AutoOffsetResetStrategy.valueOf("invalid"));
+        assertThrows(IllegalArgumentException.class, () -> AutoOffsetResetStrategy.valueOf("LATEST"));
+        assertThrows(IllegalArgumentException.class, () -> AutoOffsetResetStrategy.valueOf(""));


Should we add a test case for null?

mjsax · 2024-11-20T20:33:37Z

...s/src/test/java/org/apache/kafka/clients/consumer/internals/AutoOffsetResetStrategyTest.java

+        assertDoesNotThrow(() -> validator.ensureValid("test", "none"));
+        assertThrows(ConfigException.class, () -> validator.ensureValid("test", "invalid"));
+        assertThrows(ConfigException.class, () -> validator.ensureValid("test", "LATEST"));
+        assertThrows(ConfigException.class, () -> validator.ensureValid("test", ""));


mjsax · 2024-11-20T20:40:57Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AutoOffsetResetStrategy.java

+
+    public static final AutoOffsetResetStrategy EARLIEST = new AutoOffsetResetStrategy(EARLIEST_STRATEGY_NAME);
+    public static final AutoOffsetResetStrategy LATEST = new AutoOffsetResetStrategy(LATEST_STRATEGY_NAME);
+    public static final AutoOffsetResetStrategy NONE = new AutoOffsetResetStrategy(NONE_STRATEGY_NAME);


If we use an enum we don't need this boiler plate objects?

These are not boilerplate objects. These are static instances to represent current strategies.
The main reason for converting enum into class is to allow us to add additional reset strategies with some properties.

For example, we will be adding new class like below to represent duration based reset strategy:

DurationBasedOffsetResetStrategy extends AutoOffsetResetStrategy { DurationBasedOffsetResetStrategy(String name, String isoDuration) { } }

mjsax

Personally, I would prefer enum plus Optional<Duration> instead of static final object, but guess it works either way.

omkreddy · 2024-11-23T04:49:23Z

Personally, I would prefer enum plus Optional<Duration> instead of static final object, but guess it works either way.

I have moved the sting names to enum now. I will keep static instances as they easily fit into current code.

AndrewJSchofield

Thanks for the PR. Just a few minor comments.

AndrewJSchofield · 2024-11-25T09:46:50Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

+                } else if (offsetResetStrategy == AutoOffsetResetStrategy.LATEST) {
+                    addToResetList(partition, seekToEnd, "Setting topic '{}' to consume from {} offset", "latest", loggedTopics);
+                } else if (offsetResetStrategy == AutoOffsetResetStrategy.NONE) {
+                    if (AutoOffsetResetStrategy.EARLIEST == AutoOffsetResetStrategy.fromString(originalReset)) {


I suggest making a variable to hold AutoOffsetResetStrategy.fromString(originalReset) so this code is not repeated in both if conditions.

AndrewJSchofield · 2024-11-25T09:54:30Z

clients/src/main/java/org/apache/kafka/clients/consumer/OffsetResetStrategy.java

@@ -18,6 +18,10 @@

 import java.util.Locale;

+/**
+ * @deprecated Since 4.0.


I suggest adding an indication about what to migrate to instead.

omkreddy · 2024-11-25T10:21:48Z

@AndrewJSchofield Thanks for the review. Updated the PR,

AndrewJSchofield · 2024-11-25T10:53:36Z

clients/src/main/java/org/apache/kafka/clients/consumer/OffsetResetStrategy.java

@@ -19,7 +19,7 @@
 import java.util.Locale;

 /**
- * @deprecated Since 4.0.
+ * @deprecated Since 4.0. Use {@link org.apache.kafka.clients.consumer.internals.AutoOffsetResetStrategy instead.


nit: Missing the closing '}'.

- Deprecates OffsetResetStrategy enum - Adds new internal class AutoOffsetResetStrategy - Replaces all OffsetResetStrategy enum usages with AutoOffsetResetStrategy - Deprecate old/Add new constructors to MockConsumer Reviewers: Andrew Schofield <[email protected]>, Matthias J. Sax <[email protected]>

omkreddy requested review from AndrewJSchofield, lianetm and mjsax November 19, 2024 12:14

github-actions bot added streams consumer tools connect performance storage Pull requests that target the storage module clients labels Nov 19, 2024

omkreddy force-pushed the KAFKA-18013-consumer branch from 5fb0ba3 to ab1a138 Compare November 19, 2024 15:19

mjsax reviewed Nov 20, 2024

View reviewed changes

omkreddy force-pushed the KAFKA-18013-consumer branch 2 times, most recently from 2d36c40 to f4c60ea Compare November 22, 2024 03:06

mjsax approved these changes Nov 22, 2024

View reviewed changes

omkreddy force-pushed the KAFKA-18013-consumer branch from f4c60ea to a71c5f0 Compare November 23, 2024 03:46

omkreddy added 7 commits November 25, 2024 12:32

KAFKA-18013: Add AutoOffsetResetStrategy internal class

04ca402

implement toString() method

ec92ca8

Address review comments

ebcddc0

cleanup

0f8e16c

fix conflcts

8a02ac8

Add enum

0f8547a

fix spotless

d336dfd

omkreddy force-pushed the KAFKA-18013-consumer branch from 78c7892 to d336dfd Compare November 25, 2024 07:18

AndrewJSchofield requested changes Nov 25, 2024

View reviewed changes

Add review comments

aeb2c6e

AndrewJSchofield requested changes Nov 25, 2024

View reviewed changes

Fix comment

74b194a

Fix checkstyle

4f58c52

omkreddy merged commit 3268435 into apache:trunk Nov 25, 2024
8 checks passed

peterxcli mentioned this pull request Dec 8, 2024

KAFKA-18015: Add support for duration based offset reset strategy to Kafka Streams #17973

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-18013: Add AutoOffsetResetStrategy internal class #17858

KAFKA-18013: Add AutoOffsetResetStrategy internal class #17858

omkreddy commented Nov 19, 2024 •

edited

Loading

mjsax left a comment

mjsax Nov 20, 2024

omkreddy Nov 21, 2024 •

edited

Loading

mjsax Nov 20, 2024

omkreddy Nov 21, 2024 •

edited

Loading

mjsax Nov 22, 2024

mjsax Nov 20, 2024

mjsax Nov 20, 2024

mjsax Nov 20, 2024

mjsax Nov 20, 2024

omkreddy Nov 21, 2024

mjsax left a comment

omkreddy commented Nov 23, 2024

AndrewJSchofield left a comment

AndrewJSchofield Nov 25, 2024

AndrewJSchofield Nov 25, 2024

omkreddy commented Nov 25, 2024

AndrewJSchofield Nov 25, 2024

KAFKA-18013: Add AutoOffsetResetStrategy internal class #17858

KAFKA-18013: Add AutoOffsetResetStrategy internal class #17858

Conversation

omkreddy commented Nov 19, 2024 • edited Loading

mjsax left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

omkreddy Nov 21, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

omkreddy Nov 21, 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

mjsax left a comment

Choose a reason for hiding this comment

omkreddy commented Nov 23, 2024

AndrewJSchofield left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

omkreddy commented Nov 25, 2024

Choose a reason for hiding this comment

omkreddy commented Nov 19, 2024 •

edited

Loading

omkreddy Nov 21, 2024 •

edited

Loading

omkreddy Nov 21, 2024 •

edited

Loading