SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DltAwareProcessor improvements

Browse Source 
- Instead of using a BiFunction as a delegate, use standard Function that takes the full record
 - Remove Supplier<Long> that was used to handle record time stamps since this is no longer needed
 - Docs cleanup
This commit is contained in:
Soby Chacko
2023-09-26 21:32:30 -04:00
parent ca2d3046b9
commit bca3dd5566
3 changed files with 16 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/DltAwareProcessor.java
View File

@@ -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<KIn, VIn, KOut, VOut> implements Processor<KIn, VIn, KOut, VOut> {
/**
* Delegate {@link BiFunction} that is responsible for processing the data.
* Delegate {@link Function} that is responsible for processing the data.
*/
private final BiFunction<KIn, VIn, KeyValue<KOut, VOut>> delegateFunction;
/**
* Event time for the forwarded downstream record.
*/
private final Supplier<Long> recordTimeSupplier;
private final Function<Record<KIn, VIn>, Record<KOut, VOut>> delegateFunction;
/**
* DLT destination.
@@ -70,26 +63,13 @@ public class DltAwareProcessor<KIn, VIn, KOut, VOut> implements Processor<KIn, V
/**
*
* @param delegateFunction {@link BiFunction} to process the data
* @param delegateFunction {@link Function} to process the data
* @param dltDestination DLT destination
* @param dltPublishingContext {@link DltPublishingContext}
*/
public DltAwareProcessor(BiFunction<KIn, VIn, KeyValue<KOut, VOut>> delegateFunction, String dltDestination,
public DltAwareProcessor(Function<Record<KIn, VIn>, Record<KOut, VOut>> 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<KIn, VIn, KeyValue<KOut, VOut>> delegateFunction, String dltDestination,
DltPublishingContext dltPublishingContext, Supplier<Long> 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<KIn, VIn, KOut, VOut> implements Processor<KIn, V
/**
*
* @param delegateFunction {@link BiFunction} to process the data
* @param delegateFunction {@link Function} to process the data
* @param processorRecordRecoverer {@link BiConsumer} that recovers failed records
*/
public DltAwareProcessor(BiFunction<KIn, VIn, KeyValue<KOut, VOut>> delegateFunction,
public DltAwareProcessor(Function<Record<KIn, VIn>, Record<KOut, VOut>> delegateFunction,
BiConsumer<Record<KIn, VIn>, 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<KIn, VIn, KeyValue<KOut, VOut>> delegateFunction,
BiConsumer<Record<KIn, VIn>, Exception> processorRecordRecoverer, Supplier<Long> 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<KIn, VIn, KOut, VOut> implements Processor<KIn, V
@Override
public void process(Record<KIn, VIn> record) {
try {
KeyValue<KOut, VOut> keyValue = this.delegateFunction.apply(record.key(), record.value());
Record<KOut, VOut> downstreamRecord = new Record<>(keyValue.key, keyValue.value, recordTimeSupplier.get(), record.headers());
Record<KOut, VOut> downstreamRecord = this.delegateFunction.apply(record);
this.context.forward(downstreamRecord);
}
catch (Exception exception) {

2
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DltAwareProcessorTests.java
View File

@@ -112,7 +112,7 @@ public class DltAwareProcessorTests {
@Bean
public java.util.function.Consumer<KStream<String, String>> errorStream(DltPublishingContext dltSenderContext) {
return input -> input
.process(() -> new DltAwareProcessor<>((k, v) -> {
.process(() -> new DltAwareProcessor<>(rec -> {
throw new RuntimeException("error");
}, "hello-dlt-1", dltSenderContext));
}

18
docs/modules/ROOT/pages/kafka/kafka-streams-binder/error-handling.adoc
View File

@@ -158,7 +158,7 @@ Here is how you can do that.
@Bean
public java.util.function.Function<KStream<String, String>, KStream<String, String>> 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<String, String>, KStream<String, Stri
The business logic code from the original `map` call now has been moved as part of `KStream#process` method call, which takes a `ProcessorSupplier`.
We, then, pass in the custom `DltAwareProcessor,` which is capable to publishing to a DLT.
The constructor for `DltAwareProcessor` above takes three parameters - a `BiFunction` that takes the key and value of the input record and then the business logic operation as part of the `BiFunction` body, the DLT topic, and finally a `DltPublishingContext`. When the `BiFunction`'s lambda expression throws an exception, the `DltAwareProcessor` will send the input record to a DLT. The `DltPublishingContext` provides `DltAwareProcessor` the necessary publishing infrastructure beans.
The constructor for `DltAwareProcessor` above takes three parameters - a `Function` that takes the input record and then the business logic operation as part of the `Function` body, the DLT topic, and finally a `DltPublishingContext`. When the `Function`'s lambda expression throws an exception, the `DltAwareProcessor` will send the input record to a DLT. The `DltPublishingContext` provides `DltAwareProcessor` the necessary publishing infrastructure beans.
The `DltPublishingContext` is autoconfigured by the binder, so that you can inject directly this into the application.
If you want to provide a custom timestamp on the record that gets published to the DLT, then you can provide an optional fourth constructor argument which is a `Supplier<Long>`.
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<String, String>, KStream<String, String>> 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<String, String>, KStream<String, Stri
```
In this case, when the record fails, the `DltAwareProcessor`, instead of using its built-in recoverer which publishes to a DLT, uses the user provided recoverer which is a `BiConsumer` that takes the failed record and the exception thrown as arguments.
In this case also, you can provide an optional `Supplier<Long>` to dictate the timestamp used in the record passed in to the `BiConsumer` recoverer.