Adding config auto.offset.reset to PulsarKafkaConsumer #3273

ConcurrencyPractitioner · 2019-01-02T04:10:33Z

Motivation

Currently, in KafkaConsumer, there is a new config which is called auto.offset.reset. It is currently ignored in PulsarKafkaConsumer, so we wish to implement it.

Modifications

We added some new code which defined the behavior if auto.offset.reset == "earliest" or "latest".

Result

It would automatically seek to the last offset or the first offset depending on the set config.

...pulsar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java

sijie · 2019-01-02T08:52:46Z

...pulsar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java

@@ -159,6 +163,16 @@ private PulsarKafkaConsumer(ConsumerConfig config, Deserializer<K> keyDeserializ
        }
    }

+    private OffsetResetStrategy getStrategy(final String strategy) {
+    	if (strategy.equals("earliest")) {


couldn't strategy be null?

Yeah, will change to the logic to take that into account.

sijie · 2019-01-02T08:53:40Z

...pulsar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java

+    		return OffsetResetStrategy.EARLIEST;
+    	} else if (strategy.equals("latest")) {
+    		return OffsetResetStrategy.LATEST;
+    	} else {


I would prefer failing-fast if strategy is not null or empty here and logging a message to tell users that this is a wrong setting.

Sure, will do.

sijie · 2019-01-02T08:54:10Z

...pulsar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java

@@ -311,6 +325,9 @@ public void unsubscribe() {
                long offset = MessageIdUtils.getOffset(msgId);

                TopicPartition tp = new TopicPartition(topic, partition);
+                if (lastReceivedOffset.get(tp) == null) {


can you explain why you are adding this logic here?

Well, I noticed that lastReceivedOffset is cleared whenever a seek-to-end or seek-to-front is called. So looking into it, I found that whenever lastReceivedOffset.get(tp) returns null, it would indicate the current offset position is invalid (or could not be found since we are using a map). Taking advantage of this fact, I would just call seekToEnd or seekToFront to achieve the same behavior as KafkaConsumer. Because in KafkaConsumer internals, their position() method and seekToEnd() and seekToStart() uses the same approach.

…rg/apache/kafka/clients/consumer/PulsarKafkaConsumer.java Co-Authored-By: ConcurrencyPractitioner <yohan.richard.yu@gmail.com>

…ulsar

asfgit · 2019-01-02T17:32:30Z

SUCCESS

--none--

merlimat · 2019-01-02T17:57:33Z

...pulsar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java

@@ -65,6 +66,7 @@

 @Slf4j
 public class PulsarKafkaConsumer<K, V> implements Consumer<K, V>, MessageListener<byte[]> {
+    private static enum OffsetResetStrategy {EARLIEST, LATEST, NONE}


Why not using directly SubscriptionInitialPosition?

Oh, never thought about that. I would take a look. :)

Well, I used SubscriptionInitialPosition later. We would still need to seek to the start or the end depending on the config setting. @merlimat Do you want me to get rid of the enum?

merlimat

Documentation on the website should also be updated to reflect the newly supported feature

ConcurrencyPractitioner · 2019-01-02T20:18:53Z

One more thing. @merlimat Do you know what file to change for the documentation.

merlimat · 2019-01-02T20:27:52Z

@ConcurrencyPractitioner In site2/docs/adaptors-kafka.md there's a table with the compatiblity matrix.

ConcurrencyPractitioner · 2019-01-02T21:19:52Z

Ok @merlimat Removed the enum. Do you have any other changes you want seen done?

merlimat · 2019-01-02T21:23:13Z

...pulsar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java

@@ -143,6 +146,7 @@ private PulsarKafkaConsumer(ConsumerConfig config, Deserializer<K> keyDeserializ

        groupId = config.getString(ConsumerConfig.GROUP_ID_CONFIG);
        isAutoCommit = config.getBoolean(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG);
+        strategy = getStrategy(config.getString(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG));


Wouldn't this give error if config doesn't include the specified key? Shouldn't we have a default there?

In Kafka the default is latest.

Oh. I will take that into account.

merlimat · 2019-01-02T21:23:31Z

...pulsar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java

@@ -159,6 +163,17 @@ private PulsarKafkaConsumer(ConsumerConfig config, Deserializer<K> keyDeserializ
        }
    }

+    private SubscriptionInitialPosition getStrategy(final String strategy) {
+    	switch(strategy) {
+    		case "earliest":


Use 4 spaces for indentation

merlimat · 2019-01-02T21:24:47Z

...pulsar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java

@@ -492,9 +510,21 @@ public void seekToEnd(Collection<TopicPartition> partitions) {
    @Override
    public long position(TopicPartition partition) {
        Long offset = lastReceivedOffset.get(partition);
-        return offset != null ? offset : -1l;
+        if (offset == null) {
+        	return resetOffsets(partition).getValue();


4 spaces indent

ConcurrencyPractitioner · 2019-01-02T21:38:21Z

Ok, done. @merlimat

ConcurrencyPractitioner · 2019-01-02T23:16:23Z

@sijie Do you think there is anything else I need?

merlimat · 2019-01-03T03:51:09Z

site2/docs/adaptors-kafka.md

@@ -129,6 +129,7 @@ Properties:
 | Config property                         | Supported | Notes                                                                         |
 |:----------------------------------------|:----------|:------------------------------------------------------------------------------|
 | `acks`                                  | Ignored   | Durability and quorum writes are configured at the namespace level            |
+| `auto.offset.reset`			  | Yes       | Will throw an exception if this property is not defined (i.e. is none).	      |


The note is no longer accurate now

merlimat · 2019-01-03T03:51:56Z

...pulsar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java

    }

+    private SubscriptionInitialPosition resetOffsets(final TopicPartition partition) {
+    	if (strategy == SubscriptionInitialPosition.Earliest) {
+    		seekToBeginning(Collections.singleton(partition));


There are still several cases with tabs instead of spaces

ConcurrencyPractitioner · 2019-01-03T21:51:06Z

Hi @merlimat Updated the PR. Hope this wraps things up.

sijie · 2019-01-03T23:28:57Z

...pulsar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java

+    	    case "earliest":
+    	        return SubscriptionInitialPosition.Earliest;
+    	    default:
+    		    return SubscriptionInitialPosition.Latest;


wrong indent?

Yeah, these things happen. Oh, some tests were broken by my change. We should fix that before merging.

Actually, I discovered a few things. When I looked at the failing tests for integration, KafkaApiTest failed. But looking in detail, I found that we were using KafkaConsumer instead of PulsarKafkaConsumer. I don't think the test failures are related to my change. But I'm not too sure. Am I right?

PulsarKafkaConsumer was shaded to KafkaConsumer. so if there are test failures, it should be related.

ConcurrencyPractitioner · 2019-01-03T23:38:33Z

Also, I did some digging. I couldn't seem to find any tests for PulsarKafkaConsumer. Should we add some tests to confirm the method's behavior? I found this to be good practice. WDYT?

sijie · 2019-01-03T23:47:26Z

Should we add some tests to confirm the method's behavior? I found this to be good practice. WDYT?

In general, it is a good practice to add tests (unit test at least, integration tests would be great) to make sure newly introduced code changes or features are covered by tests. so any subsequent changes can be verified using those tests in future.

ConcurrencyPractitioner · 2019-01-05T01:50:43Z

Ok, fixed the test. I will probably add a new one and this should be good to go.

merlimat

👍

ConcurrencyPractitioner added 2 commits January 1, 2019 20:07

Adding config auto.offset.reset to PulsarKafkaConsumer

04dfa41

Adding similar behavior to poll

3826bc1

sijie requested a review from merlimat January 2, 2019 08:50

sijie assigned ConcurrencyPractitioner Jan 2, 2019

sijie added area/client type/feature The PR added a new feature or issue requested a new feature labels Jan 2, 2019

sijie added this to the 2.3.0 milestone Jan 2, 2019

sijie requested changes Jan 2, 2019

View reviewed changes

sijie and others added 4 commits January 2, 2019 11:11

Update pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/o…

f852ef1

…rg/apache/kafka/clients/consumer/PulsarKafkaConsumer.java Co-Authored-By: ConcurrencyPractitioner <yohan.richard.yu@gmail.com>

Adding some changes

84f2097

Merge branch 'master' of https://github.com/ConcurrencyPractitioner/p…

7b535bd

…ulsar

Fixing indentation

f3190a1

merlimat reviewed Jan 2, 2019

View reviewed changes

Modifying postion method

f7cb5be

ConcurrencyPractitioner added 2 commits January 2, 2019 13:10

Adding entry to compatibility matridx

6f25679

Removing enum

81b84af

merlimat reviewed Jan 2, 2019

View reviewed changes

Fixing indentation and policy

8a40730

merlimat reviewed Jan 3, 2019

View reviewed changes

removing indents

9a99c8b

sijie approved these changes Jan 3, 2019

View reviewed changes

Fixing indent

84f9b80

ConcurrencyPractitioner added 5 commits January 3, 2019 18:06

Adding log statements

65ddcff

Modifying seekToEnd behavior

d46ed29

Adding check

e36a355

updating position

08c717d

Reverting accidentaldeletion

596b783

merlimat approved these changes Jan 5, 2019

View reviewed changes

merlimat merged commit ef82f04 into apache:master Jan 5, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Adding config auto.offset.reset to PulsarKafkaConsumer #3273

Adding config auto.offset.reset to PulsarKafkaConsumer #3273

ConcurrencyPractitioner commented Jan 2, 2019

sijie Jan 2, 2019

ConcurrencyPractitioner Jan 2, 2019

sijie Jan 2, 2019

ConcurrencyPractitioner Jan 2, 2019

sijie Jan 2, 2019

ConcurrencyPractitioner Jan 2, 2019

asfgit commented Jan 2, 2019

merlimat Jan 2, 2019

ConcurrencyPractitioner Jan 2, 2019

ConcurrencyPractitioner Jan 2, 2019

merlimat left a comment

ConcurrencyPractitioner commented Jan 2, 2019

merlimat commented Jan 2, 2019

ConcurrencyPractitioner commented Jan 2, 2019

merlimat Jan 2, 2019

merlimat Jan 2, 2019

ConcurrencyPractitioner Jan 2, 2019

merlimat Jan 2, 2019

ConcurrencyPractitioner Jan 2, 2019

merlimat Jan 2, 2019

ConcurrencyPractitioner commented Jan 2, 2019 •

edited

Loading

ConcurrencyPractitioner commented Jan 2, 2019

merlimat Jan 3, 2019

merlimat Jan 3, 2019

ConcurrencyPractitioner commented Jan 3, 2019

sijie Jan 3, 2019

ConcurrencyPractitioner Jan 3, 2019

ConcurrencyPractitioner Jan 3, 2019

sijie Jan 3, 2019

ConcurrencyPractitioner commented Jan 3, 2019

sijie commented Jan 3, 2019

ConcurrencyPractitioner commented Jan 5, 2019

merlimat left a comment

Adding config auto.offset.reset to PulsarKafkaConsumer #3273

Adding config auto.offset.reset to PulsarKafkaConsumer #3273

Conversation

ConcurrencyPractitioner commented Jan 2, 2019

Motivation

Modifications

Result

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

asfgit commented Jan 2, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

merlimat left a comment

Choose a reason for hiding this comment

ConcurrencyPractitioner commented Jan 2, 2019

merlimat commented Jan 2, 2019

ConcurrencyPractitioner commented Jan 2, 2019

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

ConcurrencyPractitioner commented Jan 2, 2019 • edited Loading

ConcurrencyPractitioner commented Jan 2, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ConcurrencyPractitioner commented Jan 3, 2019

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

ConcurrencyPractitioner commented Jan 3, 2019

sijie commented Jan 3, 2019

ConcurrencyPractitioner commented Jan 5, 2019

merlimat left a comment

Choose a reason for hiding this comment

ConcurrencyPractitioner commented Jan 2, 2019 •

edited

Loading