GH-129: Implement Consumer Seek #173

garyrussell · 2016-08-23T20:52:42Z

Resolves #129

TODO: docs, pass callback through adapters.

stepio · 2016-08-23T21:17:54Z

...ng-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

-			if (theListener instanceof MessageListener) {
-				this.listener = (MessageListener<K, V>) theListener;
+			if (this.theListener instanceof MessageListener) {
+				this.listener = (MessageListener<K, V>) this.theListener;
 				this.batchListener = null;


As I see, this code is a part of constructor. In this case initializing object's field with null is unnecessary - java does this. So now it's just extra code, which makes the method longer and less readable.

That is incorrect - since the field is marked final, the compiler requires it to be explicitly initialized.

Good point, missed final.

artembilan · 2016-08-25T21:20:45Z

Must be rebased to the latest master.

artembilan · 2016-08-25T21:23:58Z

...ng-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -896,6 +921,11 @@ private void commitIfNecessary() {
 			}
 		}

+		@Override
+		public void seek(String topic, int partition, long offset) {
+			this.seeks.add(new TopicPartitionInitialOffset(topic, partition, offset));


I wonder why we can't reach the same via just top-level container operation.
Moreover expose it to JMX.
What am I missing?
Thanks

Because it somehow needs to be directed to the right consumer.

KafkaMessageListenerContainer has 1-1 relationship with the ListenerConsumer. So, I believe, the question is closed here.

The ConcurrentMessageListenerContainer builds a set of containers based on the partitionSubset(containerProperties, i). Together with the Collection<TopicPartition> getAssignedPartitions(), it seems for me we can reach a uniqueness and correctly assign requested seek to the appropriate KafkaMessageListenerContainer.
No ?

My concern is that the topics/partitions can move when the broker does a rebalance of the assignment - I want to be sure that the seek is done in the context of the current state at the time the listener is called. That way, if the consumer no longer owns that partition the seek operation will fail.

artembilan · 2016-08-25T21:39:57Z

The solution looks cool, but I'm curious what drove you away of the simple seek() on the AbstractMessageListenerContainer directly.

With good valuable argument I can merge it as is or let you go ahead with Docs 😄
Thanks

artembilan · 2016-08-25T21:44:20Z

...ng-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

 			}
 			Assert.state(!this.isBatchListener || !this.isRecordAck, "Cannot use AckMode.RECORD with a batch listener");
+			if (this.autoCommit && this.theListener instanceof ConsumerSeekAware) {
+				((ConsumerSeekAware) this.theListener).registerSeekCallback(this);


Why do we need this one if we registerSeekCallback() any way in the ListenerInvoker.run() ?

Because with autoCommit = true, the listener is invoked on the consumer thread.

I wasn't able to step in here in debug mode during testing. Maybe we just don't have any test on the matter?..

Having your condition, "the listener is invoked on the consumer thread." and "listeners should store the callback in a {@code ThreadLocal}.", plus the test-case in the EnableKafkaIntegrationTests with exactly a ThreadLocal, I dug further and found this code in the KafkaMessageListenerContainer.doStart():

this.listenerConsumer = new ListenerConsumer(this.listener, this.acknowledgingMessageListener); setRunning(true); this.listenerConsumerFuture = containerProperties .getConsumerTaskExecutor() .submitListenable(this.listenerConsumer);

So, we create ListenerConsumer instance in the KafkaMessageListenerContainer.doStart() Thread, but run() it in a different Thread.

My concern that this .registerSeekCallback(this) for the autoCommit = true is done in a wrong Thread. Therefore I won't be able to get a proper ThreadLocal value, when I consume messages.

Am I missing anything?
Thanks

Yes, we don't have a test case; will add one - good catch - I had a similar problem with the other case - I called from the ctor instead of the run method in the ListenerInvoker. See the second commit where I moved that code.

garyrussell · 2016-08-29T14:18:01Z

Hmmm - no travis check - weird.

artembilan · 2016-08-29T14:31:25Z

no travis check

This branch has conflicts that must be resolved

Can't that be a reason?

artembilan · 2016-08-29T15:12:26Z

...src/main/java/org/springframework/kafka/listener/adapter/AbstractMessageListenerAdapter.java

+ * @param <T> the delegate type.
+ *
+ * @author Gary Russell
+ * @since 5.0


artembilan · 2016-08-29T17:49:56Z

Fine with me.

Do you still have something else to do here, e.g. Docs?
Otherwise I can merge.

garyrussell · 2016-08-29T21:14:50Z

Will add a docs commit tomorrow.

artembilan · 2016-08-30T17:18:06Z

spring-kafka/src/main/java/org/springframework/kafka/support/TopicPartitionCurrentOffset.java

+/**
+ * Identifies the current position of a TopicPartition.
+ *
+ * @author Artem Bilan


This is your class and @since 1.1
😄

Although I still try to understand if we need it...
Maybe simple Map<TopicPartition, Long> could be enough?

Or maybe we can just rename TopicPartitionInitialOffset to TopicPartitionOffset...

I changed it to a Map.

artembilan · 2016-08-30T17:40:05Z

I think we need What's New note also.

artembilan · 2016-08-30T17:40:32Z

Rebase is needed as well.

artembilan · 2016-08-30T18:02:10Z

That's all for this round.

Resolves spring-projects#129

garyrussell · 2016-08-30T18:05:23Z

Needs rebase

- support seek from a `@KafkaListener` - pass the callback through the adapters - set the callback on the correct thread when not auto-commit

Also rebased.

artembilan · 2016-08-30T19:13:07Z

I think I'm fine with the fix.

Polling locally for the last review round and probably merge...

artembilan · 2016-08-30T19:14:59Z

spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerSeekAware.java

+
+	/**
+	 * Register the callback to use when seeking at some arbitrary time. When used with a
+	 * {@code ConcurrentMessageListenerContainer} or the same listener instance in multupe


Typo: multiple

artembilan · 2016-08-30T20:21:15Z

Merged as b18ef3e

stepio reviewed Aug 23, 2016
View reviewed changes

artembilan reviewed Aug 25, 2016
View reviewed changes

garyrussell force-pushed the GH-129 branch 2 times, most recently from 1413897 to e6be94d Compare August 29, 2016 14:15

artembilan reviewed Aug 29, 2016
View reviewed changes

garyrussell force-pushed the GH-129 branch from e6be94d to da803a9 Compare August 29, 2016 15:21

artembilan reviewed Aug 30, 2016
View reviewed changes

spring-projectsGH-129: Implement Consumer Seek

97248b5

Resolves spring-projects#129

garyrussell added 4 commits August 30, 2016 14:11

Polishing - @KafkaListener Seeks and More

13b71db

- support seek from a `@KafkaListener` - pass the callback through the adapters - set the callback on the correct thread when not auto-commit

Polishing - PR Comments

b5025f2

Fix Threading with auto-commit

0b86c66

Also rebased.

Fix since

ee094fe

garyrussell added 3 commits August 30, 2016 14:16

Docs + Initial Seek for Assigned Partitions

135a444

Add Seek on Idle Container Detection

1cec928

Doc Polish and Rebase

220accb

garyrussell force-pushed the GH-129 branch from 860d988 to 220accb Compare August 30, 2016 18:39

artembilan reviewed Aug 30, 2016
View reviewed changes

artembilan closed this Aug 30, 2016

garyrussell deleted the GH-129 branch March 30, 2018 21:49

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

GH-129: Implement Consumer Seek #173

GH-129: Implement Consumer Seek #173

garyrussell commented Aug 23, 2016

stepio Aug 23, 2016

garyrussell Aug 23, 2016

stepio Aug 23, 2016

artembilan commented Aug 25, 2016

artembilan Aug 25, 2016

garyrussell Aug 25, 2016

artembilan Aug 25, 2016

garyrussell Aug 25, 2016

artembilan commented Aug 25, 2016

artembilan Aug 25, 2016

garyrussell Aug 25, 2016

artembilan Aug 29, 2016

garyrussell Aug 29, 2016

garyrussell commented Aug 29, 2016

artembilan commented Aug 29, 2016

artembilan Aug 29, 2016

artembilan commented Aug 29, 2016

garyrussell commented Aug 29, 2016

artembilan Aug 30, 2016

artembilan Aug 30, 2016

garyrussell Aug 30, 2016

artembilan commented Aug 30, 2016

artembilan commented Aug 30, 2016

artembilan commented Aug 30, 2016

garyrussell commented Aug 30, 2016

artembilan commented Aug 30, 2016

artembilan Aug 30, 2016

artembilan commented Aug 30, 2016

GH-129: Implement Consumer Seek #173

GH-129: Implement Consumer Seek #173

Conversation

garyrussell commented Aug 23, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

artembilan commented Aug 25, 2016

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

artembilan commented Aug 25, 2016

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

garyrussell commented Aug 29, 2016

artembilan commented Aug 29, 2016

Choose a reason for hiding this comment

artembilan commented Aug 29, 2016

garyrussell commented Aug 29, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

artembilan commented Aug 30, 2016

artembilan commented Aug 30, 2016

artembilan commented Aug 30, 2016

garyrussell commented Aug 30, 2016

artembilan commented Aug 30, 2016

Choose a reason for hiding this comment

artembilan commented Aug 30, 2016