KafkaTemplate Request/Reply #544

garyrussell · 2018-01-26T16:25:16Z

Add AsyncKafkaTemplate with sendAndReceive()
In @KafkaListener, echo the correlationId header

Needs docs
I marked it @since 2.1.3 because we need to build 2.1.2 on Monday an I don't think this will be merged by then.

artembilan · 2018-01-26T16:36:22Z

Something is not stable:

org.springframework.kafka.AsyncKafkaTemplateTests > testGood FAILED
    java.util.concurrent.TimeoutException at AsyncKafkaTemplateTests.java:94

See Travis report.

Why don't deffer this until 2.2 then though?

garyrussell · 2018-01-26T19:04:12Z

We don't have any immediate plans for a 2.2 release. Why would we delay?

artembilan · 2018-01-26T19:11:24Z

SF 5.1, Reactor 3.2? Might be Kafka 1.1...

OK. Never mind. I'm just a bit nervous to introduce new features in the point release. More versions support in the future when we face a bug in that feature.

artembilan

That's all.
Need JavaDocs and Docs.

artembilan · 2018-01-30T22:07:24Z

spring-kafka/src/main/java/org/springframework/kafka/AsyncKafkaTemplate.java

+			GenericMessageListenerContainer<K, R> replyContainer, boolean autoFlush) {
+		super(producerFactory, autoFlush);
+		this.replyContainer = replyContainer;
+		this.replyContainer.setupMessageListener(this);


Assert.notNull(replyContainer);

artembilan · 2018-01-30T22:08:47Z

spring-kafka/src/main/java/org/springframework/kafka/AsyncKafkaTemplate.java

+
+	@Override
+	public void afterPropertiesSet() throws Exception {
+		Assert.notNull(this.scheduler, "A scheduler is required for timeouts");


Why just don't create an internal instance then?

It doesn't look correct to allow to create this AsyncKafkaTemplate without TaskScheduler, but then require it later.

artembilan · 2018-01-30T22:15:28Z

spring-kafka/src/main/java/org/springframework/kafka/AsyncKafkaTemplate.java

+	public RequestReplyFuture<K, V, R> sendAndReceive(ProducerRecord<K, V> record) {
+		String correlationId = createCorrelationId();
+		record.headers()
+				.add(new RecordHeader(KafkaHeaders.CORRELATION_ID, correlationId.getBytes(StandardCharsets.UTF_8)));


Looks like we need exactly byte[] for the value, so, let's make that createCorrelationId() to return byte[] for us!

The default one indeed can be what you do with UUID, but as this:

ByteBuffer bb = ByteBuffer.wrap(new byte[16]); bb.putLong(uuid.getMostSignificantBits()); bb.putLong(uuid.getLeastSignificantBits()); return bb.array();

This way we will be free of the String to byte[] conversion here.

WDYT?

artembilan · 2018-01-30T22:18:25Z

spring-kafka/src/main/java/org/springframework/kafka/AsyncKafkaTemplate.java

+		}
+		RequestReplyFuture<K, V, R> future = new RequestReplyFuture<>();
+		this.futures.put(correlationId, future);
+		future.setSendFuture(send(record));


What if we couldn't send here?
Can't we perform this.futures.put() already after this send()?
Or do you mean there might be a race condition that reply will come earlier than we perform this.futures.put() ?

Thanks

Yes (race condition).

OK. So, the next schedule must be in the final block for the case when we got an exception from the send(record). Or even better do not schedule at all, but just remove in the catch. Just because we haven't sent record therefore we won't anticipate any reply at all.

Does it make sense ?

artembilan · 2018-01-30T22:55:06Z

spring-kafka/src/test/resources/log4j.properties

@@ -4,6 +4,7 @@ log4j.appender.stdout=org.apache.log4j.ConsoleAppender
 log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
 log4j.appender.stdout.layout.ConversionPattern=%d{HH:mm:ss.SSS} %-5p [%t][%c] %m%n
 log4j.category.org.springframework.kafka=WARN
+log4j.category.org.springframework.kafka.AsyncKafkaTemplate=DEBUG


I hope we will remove it during merge.

artembilan

Cool feature, but I still have some concerns.

Thanks

artembilan · 2018-01-31T18:54:36Z

spring-kafka/src/main/java/org/springframework/kafka/ReplyingKafkaTemplate.java

+	}
+
+	protected void setTaskScheduler(TaskScheduler scheduler) {
+		this.scheduler = scheduler;


Assert.notNull()

artembilan · 2018-01-31T18:56:37Z

spring-kafka/src/main/java/org/springframework/kafka/ReplyingKafkaTemplate.java

+
+	@Override
+	public synchronized void start() {
+		try {


Why no if (!this.running) ?

artembilan · 2018-01-31T18:59:41Z

spring-kafka/src/main/java/org/springframework/kafka/ReplyingKafkaTemplate.java

+		}
+		TemplateRequestReplyFuture<K, V, R> future = new TemplateRequestReplyFuture<>();
+		this.futures.put(correlationId, future);
+		future.setSendFuture(send(record));


Why still no try...catch()?
In case the send(record) throws an exception we don't reach this.scheduler.schedule() and we have an orphan entry in the this.futures

artembilan · 2018-01-31T19:05:12Z

spring-kafka/src/main/java/org/springframework/kafka/ReplyingKafkaTemplate.java

+			RequestReplyFuture<K, V, R> removed = this.futures.remove(correlationId);
+			if (removed != null) {
+				if (this.logger.isWarnEnabled()) {
+					this.logger.warn("Reply timed out for: " + record + " with correlationId: " + correlationId);


When we call stop() we perform this.replyContainer.stop() therefore no any replies are going to come back to us.
And from here all the waiting this.futures will be discarded here with this misleading warning.

What I think we need to cancel the Future from this schedule() and clean up this.futures altogether in the stop() as well after stopping this.replyContainer.

WDYT?

Canceling the futures would mean we'd have to track them. I think we can just do

this.futures.replaceAll((k, v) -> null);

in stop() and then the timeouts will be no-ops.

Actually CHM doesn't support null, but clear() will do it.

artembilan · 2018-01-31T19:11:32Z

spring-kafka/src/main/java/org/springframework/kafka/ReplyingKafkaTemplate.java

+				}
+			}
+			if (correlationId == null) {
+				this.logger.error("No correlationId found in reply: " + record);


I think this not enough. Not found. And? What now ? Did we do anything wrong?
I mean let's add some explanation into the log message what is wrong in this case.

artembilan · 2018-01-31T19:12:51Z

spring-kafka/src/main/java/org/springframework/kafka/ReplyingKafkaTemplate.java

+		private volatile Integer hashCode;
+
+		public CorrelationKey(byte[] bytes) {
+			this.correlationId = bytes;


Assert.notNull() - this is public API, so end-user may decide to try to suffer of NPE 😄

artembilan · 2018-01-31T19:15:24Z

...kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java

@@ -44,7 +44,7 @@
 * @author Artem Bilan
 */
 public abstract class AbstractMessageListenerContainer<K, V>
-		implements MessageListenerContainer, BeanNameAware, ApplicationEventPublisherAware, SmartLifecycle {
+		implements GenericMessageListenerContainer<K, V>, BeanNameAware, ApplicationEventPublisherAware {


What happened to SmartLifecycle here?

Hardly ever possible, but still breaking change.

The super.super interface (MLC) extends SL. So it was always redundant here.

artembilan · 2018-01-31T19:18:30Z

src/reference/asciidoc/kafka.adoc

+RequestReplyFuture<K, V, R> sendAndReceive(ProducerRecord<K, V> record);
+----
+
+The result is a `ListenableFuture` that will asynchronously populated with the result (or an exception, for a timeout).


will be populated ?

artembilan · 2018-01-31T19:20:54Z

src/reference/asciidoc/kafka.adoc

+}
+----
+
+The `@KafkaListener` infrastructure echoes the correlation id and determines the reply topic


Period in the end of sentence. Or some text is missed yet...

garyrussell · 2018-01-31T20:06:38Z

...rc/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java

+	/*
+	 * Needed if Jackson is not on the classpath since we can't map headers in that case.
+	 */
+	private byte[] extractNativeHeader(Object source, String header) {


Don't merge yet; I have found a better solution for when Jackson is not on the CP.

artembilan

More review after pulling locally.

artembilan · 2018-01-31T20:22:47Z

spring-kafka/src/main/java/org/springframework/kafka/ReplyingKafkaTemplate.java

+ * @since 2.1.3
+ *
+ */
+public class ReplyingKafkaTemplate<K, V, R> extends KafkaTemplate<K, V> implements BatchMessageListener<K, R>,


I think we have to push this to the core package. Otherwise we have a package tangle between root and that core:

artembilan · 2018-01-31T20:25:33Z

spring-kafka/src/main/java/org/springframework/kafka/ReplyingKafkaTemplate.java

+	@Override
+	public RequestReplyFuture<K, V, R> sendAndReceive(ProducerRecord<K, V> record) {
+		Assert.state(this.running, "Template has not been start()ed"); // NOSONAR (sync)
+		CorrelationKey correlationId = createCorrelationId();


How about to propagate record to the createCorrelationId() and allow to do record-based correlation ?

artembilan · 2018-01-31T20:29:39Z

spring-kafka/src/main/java/org/springframework/kafka/ReplyingKafkaTemplate.java

+			if (correlationId == null) {
+				this.logger.error("No correlationId found in reply: " + record
+						+ " - to use request/reply semantics, the responding server must return the correlation id "
+						+ " in the '" + KafkaHeaders.CORRELATION_ID + "' header");


I assume we may think (or wait for request) about a correlationKey property when someone would like to correlate not by out-of-the-box KafkaHeaders.CORRELATION_ID. Seems for me we provide that Spring AMQP.

Right, we may consider that in the separate issue.

Yeah - enhancement.

artembilan · 2018-01-31T20:53:59Z

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

+ * @since 2.1.3
+ *
+ */
+public class RequestReplyFuture<K, V, R> extends SettableListenableFuture<ConsumerRecord<K, R>> {


Similar package tangle we have through this class.

I think this class should go now to a new package as well.

However I would prefer to just have its inner variant and that's all.

artembilan · 2018-01-31T20:54:15Z

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

+		super();
+	}
+
+	protected void setSendFuture(ListenableFuture<SendResult<K, V>> sendFuture) {


Why just don't have it as public ?

I don't want the user to be able to set it.

Yes, good point! Why do we need this class on the top level at all?
What doesn't work keeping it as inner one?

Thanks

Where do you see a tangle? S101 doesn't see it. We can't use an inner class because then we can't have the ...Operations interface. I agree it should be moved though.

Yeah... I think there was a tangle before, when we didn't have a new requestreply.

But that's good that it is fixed now !

artembilan · 2018-01-31T21:04:11Z

...rc/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java

+	 *
+	 * @deprecated in favor of {@link #sendResponse(Object, String, Object)}.
+	 * @param result the result.
+	 * @param topic the topic.


Shoudn't @deprecated tag be in the end of JavaDocs?

<module name="AtclauseOrder"> <property name="target" value="METHOD_DEF, CTOR_DEF, VARIABLE_DEF"/> <property name="tagOrder" value="@param, @return, @throws, @since, @deprecated, @see"/> </module>

Hm... That is our Checkstyle config here. How has your change bypassed it?

artembilan · 2018-01-31T21:08:53Z

src/reference/asciidoc/kafka.adoc

+See <<annotation-send-to>> for more information about sending replies; in this case we use the default header `KafKaHeaders.REPLY_TOPIC` to indicate which topic the reply goes to.
+
+IMPORTANT: If you have multiple client instances, each will need a dedicated reply topic for each instance.
+An alternative is to set the `KafakHeaders.REPLY_PARTITION` and use a dedicated partition for each instance.


Typo: s/KafakHeaders/KafkaHeaders

artembilan · 2018-01-31T21:09:51Z

src/reference/asciidoc/whats-new.adoc

+
+==== ReplyingKafkaTemplate
+
+Starting with _version 2.1.3_ a subclass of `KafkaTemplate` is provided to support request/reply semantics.


comma before a ?

garyrussell · 2018-01-31T21:40:06Z

I'll look at the tangles tomorrow - I had to move it up to root in Spring AMQP because there was a tangle between root and core and assumed I'd have the same problem here.

artembilan · 2018-01-31T21:42:28Z

I'd have the same problem here

What I see here that we use KafkaException class in the core, but at the same time we use KafkaTemplate here in the ReplyingKafkaTemplate.

artembilan · 2018-02-01T20:26:54Z

...rc/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java

+		byte[] partitionBytes = source.getHeaders().get(KafkaHeaders.REPLY_PARTITION, byte[].class);
+		if (partitionBytes != null) {
+			String partition = new String(partitionBytes, StandardCharsets.UTF_8);
+			builder.setHeader(KafkaHeaders.PARTITION_ID, Integer.parseInt(partition));


ByteBuffer.wrap(partitionBytes).getInt() ?

artembilan

Something like that

artembilan · 2018-02-01T22:19:39Z

spring-kafka/src/main/java/org/springframework/kafka/sendreceive/ReplyingKafkaOperations.java

+ * limitations under the License.
+ */
+
+package org.springframework.kafka.sendreceive;


Why not requestreply ?

artembilan · 2018-02-01T22:20:58Z

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

+ * @since 2.1.3
+ *
+ */
+public class RequestReplyFuture<K, V, R> extends SettableListenableFuture<ConsumerRecord<K, R>> {


I think this class should go now to a new package as well.

However I would prefer to just have its inner variant and that's all.

artembilan · 2018-02-01T22:40:49Z

src/reference/asciidoc/kafka.adoc

@@ -944,6 +1053,7 @@ The `#root` object for the evaluation has 3 properties:
  - request - the inbound `ConsumerRecord` (or `ConsumerRecords` object for a batch listener))
  - source - the `org.springframework.messaging.Message<?>` converted from the `request`.
  - result - the method return result.
+- `@SendTo` (no properties) - this is treated as `!{headers['kafka_replyTopic']}` (since version _2.1.3_).


source.headers['kafka_replyTopic'] ?

artembilan · 2018-02-01T22:43:13Z

src/reference/asciidoc/kafka.adoc

@@ -987,7 +1097,7 @@ public class MultiListenerSendTo {
 ----

 When using `@SendTo`, the `ConcurrentKafkaListenerContainerFactory` must be configured with a `KafkaTemplate` in its `replyTemplate` property, to perform the send.
-Note: only the simple `send(topic, value)` method is used, so you may wish to create a subclass to generate the partition and/or key...
+NOTE: unless you are using <<replying-template,request/reply semantics>> only the simple `send(topic, value)` method is used, so you may wish to create a subclass to generate the partition and/or key...


May we fix this sentence for the ellipsis in the end?

Thanks

artembilan · 2018-02-01T22:44:29Z

spring-kafka/src/test/java/org/springframework/kafka/ReplyingKafkaTemplateTests.java

+ */
+@RunWith(SpringRunner.class)
+@DirtiesContext
+public class ReplyingKafkaTemplateTests {


Even if it is a test class I think it should go to the new package for consistency.

artembilan · 2018-02-01T23:37:07Z

Cool!

ReplyingKafkaTemplateTests.java:52:1: Redundant import from the same package - org.springframework.kafka.requestreply.ReplyingKafkaTemplate. [RedundantImport]

artembilan

Several last questions.

Will do the last run in the morning!

artembilan · 2018-02-01T23:41:58Z

spring-kafka/src/main/java/org/springframework/kafka/requestreply/ReplyingKafkaOperations.java

+ * @since 2.1.3
+ *
+ */
+public interface ReplyingKafkaOperations<K, V, R> {


May we absorb this interface and, therefore, ReplyingKafkaTemplate just to the KafkaTemplate in the future?

I think I'd rather keep it as a sub interface/class since it's configuration is more complex. Also, the RabbitTemplate has grown to 2400 LoC, which I don't want to repeat here.

artembilan · 2018-02-01T23:43:33Z

spring-kafka/src/main/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplate.java

+	 * instead but this wrapper is less expensive. We do use a BigInteger in
+	 * {@link #toString()} though.
+	 */
+	public static final class CorrelationKey {


May we have this as top-level class?

artembilan · 2018-02-01T23:49:27Z

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

+	 * The header containing a partition number on which to send the reply.
+	 * Type: binary (int) in byte[].
+	 */
+	public static final String REPLY_PARTITION = PREFIX + "replyPartition";


Should we add @since 2.1.3 to all of these new headers ?

artembilan · 2018-02-02T17:51:56Z

@garyrussell ,

ping.

I would like to hear answers to my latest comments before merging.
You might decide to fix anything else meanwhile, so I will wait.
Thanks.

garyrussell · 2018-02-02T18:43:27Z

I was waiting for your "last run" before addressing those 😄

- Add `AsyncKafkaTemplate` with `sendAndReceive()` - In `@KafkaListener`, echo the `correlationId` header

- increase timeout on good test - add exception to future on timeout

…ARTITION header.

When documenting this, we should recommend that users don't start sending until the reply partitions are assigned.

…ng since KT is also Async

artembilan

Sorry for misleading.

All good now.

Merging...

garyrussell mentioned this pull request Jan 26, 2018

Add Gateways spring-attic/spring-integration-kafka#161

Closed

garyrussell force-pushed the AsyncKT branch 7 times, most recently from 10dbddc to 1543a09 Compare January 26, 2018 18:41

garyrussell force-pushed the AsyncKT branch from e47289a to 9af3dd2 Compare January 30, 2018 20:54

artembilan requested changes Jan 30, 2018

View reviewed changes

artembilan requested changes Jan 31, 2018

View reviewed changes

garyrussell commented Jan 31, 2018

View reviewed changes

artembilan requested changes Jan 31, 2018

View reviewed changes

artembilan requested changes Feb 1, 2018

View reviewed changes

garyrussell added 6 commits February 2, 2018 13:58

KafkaTemplate Request/Reply

8a13340

- Add `AsyncKafkaTemplate` with `sendAndReceive()` - In `@KafkaListener`, echo the `correlationId` header

Polishing

ab0a665

- increase timeout on good test - add exception to future on timeout

Polishing - allow @SendTo expression to return byte[].

c77a26c

More Polishing - add standard REPLY_TOPIC header and optional REPLY_P…

fbbe768

…ARTITION header.

Improve test - wait for reply container assignment.

5599d96

When documenting this, we should recommend that users don't start sending until the reply partitions are assigned.

Polishing; docs; rename to ReplyingKafkaTemplate - Async was misleadi…

8610558

…ng since KT is also Async

garyrussell added 6 commits February 2, 2018 13:58

Polishing - PR Comments

f7b9a3a

Polishing - PR Comments and add SimpleKafkaHeaderMapper

84bbcc9

Fix package tangle - move new template to its own package.

2bca634

Use binary for REPLY_PARTITION header.

fdd5d8b

Polishing - PR comments.

a6f9c33

Rebase and final? polishing

4a393f4

garyrussell force-pushed the AsyncKT branch from 197124b to 4a393f4 Compare February 2, 2018 18:59

artembilan approved these changes Feb 2, 2018

View reviewed changes

artembilan merged commit 5adf2d1 into spring-projects:master Feb 2, 2018

garyrussell deleted the AsyncKT branch February 3, 2018 15:22


		==== ReplyingKafkaTemplate

		Starting with _version 2.1.3_ a subclass of `KafkaTemplate` is provided to support request/reply semantics.

KafkaTemplate Request/Reply #544

KafkaTemplate Request/Reply #544

Conversation

garyrussell commented Jan 26, 2018

artembilan commented Jan 26, 2018

garyrussell commented Jan 26, 2018

artembilan commented Jan 26, 2018

artembilan 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

Choose a reason for hiding this comment

artembilan 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

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

artembilan 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

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

garyrussell commented Jan 31, 2018

artembilan commented Jan 31, 2018

Choose a reason for hiding this comment

artembilan 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

artembilan commented Feb 1, 2018

artembilan 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

artembilan commented Feb 2, 2018

garyrussell commented Feb 2, 2018

artembilan left a comment

Choose a reason for hiding this comment