From bca3dd556690756a26e84d7693fc8c7e0324d1e1 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Tue, 26 Sep 2023 21:32:30 -0400 Subject: [PATCH] DltAwareProcessor improvements - Instead of using a BiFunction as a delegate, use standard Function that takes the full record - Remove Supplier that was used to handle record time stamps since this is no longer needed - Docs cleanup --- .../kafka/streams/DltAwareProcessor.java | 49 +++---------------- .../integration/DltAwareProcessorTests.java | 2 +- .../kafka-streams-binder/error-handling.adoc | 18 +++---- 3 files changed, 16 insertions(+), 53 deletions(-) diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/DltAwareProcessor.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/DltAwareProcessor.java index 5cdbbbd1a..59374cdd5 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/DltAwareProcessor.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/DltAwareProcessor.java @@ -17,10 +17,8 @@ package org.springframework.cloud.stream.binder.kafka.streams; import java.util.function.BiConsumer; -import java.util.function.BiFunction; -import java.util.function.Supplier; +import java.util.function.Function; -import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.processor.api.Processor; import org.apache.kafka.streams.processor.api.ProcessorContext; import org.apache.kafka.streams.processor.api.Record; @@ -39,14 +37,9 @@ import org.springframework.util.StringUtils; public class DltAwareProcessor implements Processor { /** - * Delegate {@link BiFunction} that is responsible for processing the data. + * Delegate {@link Function} that is responsible for processing the data. */ - private final BiFunction> delegateFunction; - - /** - * Event time for the forwarded downstream record. - */ - private final Supplier recordTimeSupplier; + private final Function, Record> delegateFunction; /** * DLT destination. @@ -70,26 +63,13 @@ public class DltAwareProcessor implements Processor> delegateFunction, String dltDestination, + public DltAwareProcessor(Function, Record> delegateFunction, String dltDestination, DltPublishingContext dltPublishingContext) { - this(delegateFunction, dltDestination, dltPublishingContext, System::currentTimeMillis); - } - - /** - * - * @param delegateFunction {@link BiFunction} to process the data - * @param dltDestination DLT destination - * @param dltPublishingContext {@link DltPublishingContext} - * @param recordTimeSupplier Supplier for downstream record timestamp - */ - public DltAwareProcessor(BiFunction> delegateFunction, String dltDestination, - DltPublishingContext dltPublishingContext, Supplier recordTimeSupplier) { this.delegateFunction = delegateFunction; - this.recordTimeSupplier = recordTimeSupplier; Assert.isTrue(StringUtils.hasText(dltDestination), "DLT Destination topic must be provided."); this.dltDestination = dltDestination; Assert.notNull(dltPublishingContext, "DltSenderContext cannot be null"); @@ -98,24 +78,12 @@ public class DltAwareProcessor implements Processor> delegateFunction, + public DltAwareProcessor(Function, Record> delegateFunction, BiConsumer, Exception> processorRecordRecoverer) { - this(delegateFunction, processorRecordRecoverer, System::currentTimeMillis); - } - - /** - * - * @param delegateFunction {@link BiFunction} to process the data - * @param processorRecordRecoverer {@link BiConsumer} that recovers failed records - * @param recordTimeSupplier Supplier for downstream record timestamp - */ - public DltAwareProcessor(BiFunction> delegateFunction, - BiConsumer, Exception> processorRecordRecoverer, Supplier recordTimeSupplier) { this.delegateFunction = delegateFunction; - this.recordTimeSupplier = recordTimeSupplier; Assert.notNull(processorRecordRecoverer, "You must provide a valid processor recoverer"); this.processorRecordRecoverer = processorRecordRecoverer; } @@ -129,8 +97,7 @@ public class DltAwareProcessor implements Processor record) { try { - KeyValue keyValue = this.delegateFunction.apply(record.key(), record.value()); - Record downstreamRecord = new Record<>(keyValue.key, keyValue.value, recordTimeSupplier.get(), record.headers()); + Record downstreamRecord = this.delegateFunction.apply(record); this.context.forward(downstreamRecord); } catch (Exception exception) { diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DltAwareProcessorTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DltAwareProcessorTests.java index 744cf9f9f..d88944c58 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DltAwareProcessorTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DltAwareProcessorTests.java @@ -112,7 +112,7 @@ public class DltAwareProcessorTests { @Bean public java.util.function.Consumer> errorStream(DltPublishingContext dltSenderContext) { return input -> input - .process(() -> new DltAwareProcessor<>((k, v) -> { + .process(() -> new DltAwareProcessor<>(rec -> { throw new RuntimeException("error"); }, "hello-dlt-1", dltSenderContext)); } diff --git a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/error-handling.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/error-handling.adoc index 0a20316c3..a983b25f1 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/error-handling.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/error-handling.adoc @@ -158,7 +158,7 @@ Here is how you can do that. @Bean public java.util.function.Function, KStream> process(DltPublishingContext dltSenderContext) { return input -> input - .process(() -> new DltAwareProcessor<>((k, v) -> { + .process(() -> new DltAwareProcessor<>(record -> { throw new RuntimeException("error"); }, "hello-dlt-1", dltPublishingContext)); } @@ -166,22 +166,20 @@ public java.util.function.Function, KStream`. -If this value is provided, then this `Supplier` is invoked each time `DltAwareProcessor` publishes to the DLT. - -If you do not want the binder to publish failed records to a DLT, then you can provide your own recoverer as a `BiConsumer`. +If you do not want the binder to publish failed records to a DLT, then you can provide your own recoverer as a `BiConsumer` that takes the input `Record` and the exception as arguments. Assume a scenario, in which you do not want to send the record to the DLT, but simply log the message and move on. -For this, it is convenient, if we can override the recovery process in `DltAwareProcessor`. -Here is an example of how you do that. +It is convenient, if we can override the default recovery mechanism provided by the `DltAwareProcessor`. + +Here is an example. ``` @Bean public java.util.function.Function, KStream> process() { return input -> input - .process(() -> new DltAwareProcessor<>((k, v) -> { + .process(() -> new DltAwareProcessor<>(record -> { throw new RuntimeException("error"); }, (record, exception) -> { @@ -191,5 +189,3 @@ public java.util.function.Function, KStream` to dictate the timestamp used in the record passed in to the `BiConsumer` recoverer. -