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

DLTAwareProcessor (Kafka Streams binder) changes

Browse Source 
- Optional BiConsumer for processor record recoverer
 - Optional Supplier for downstream record timestamp
This commit is contained in:
Soby Chacko
2023-07-24 15:02:10 -04:00
parent 25af6a76b0
commit 6bb4da89ce
3 changed files with 55 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -16,7 +16,9 @@
package org.springframework.cloud.stream.binder.kafka.streams;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Supplier;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.processor.api.Processor;
@@ -24,6 +26,8 @@ import org.apache.kafka.streams.processor.api.ProcessorContext;
import org.apache.kafka.streams.processor.api.Record;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
*
@@ -34,18 +38,44 @@ public class DltAwareProcessor<KIn, VIn, KOut, VOut> implements Processor<KIn, V
private final BiFunction<KIn, VIn, KeyValue<KOut, VOut>> delegateFunction;
private final Supplier<Long> recordTimeSupplier;
private String dltDestination;
private DltSenderContext dltSenderContext;
private BiConsumer<Record<KIn, VIn>, Exception> processorRecordRecoverer;
private ProcessorContext<KOut, VOut> context;
private final String dltDestination;
public DltAwareProcessor(BiFunction<KIn, VIn, KeyValue<KOut, VOut>> delegateFunction, String dltDestination,
DltSenderContext dltSenderContext) {
this(delegateFunction, dltDestination, dltSenderContext, System::currentTimeMillis);
}
private final DltSenderContext dltSenderContext;
public DltAwareProcessor(BiFunction<KIn, VIn, KeyValue<KOut, VOut>> businessLogic, String dltDestination, DltSenderContext dltSenderContext) {
this.delegateFunction = businessLogic;
public DltAwareProcessor(BiFunction<KIn, VIn, KeyValue<KOut, VOut>> delegateFunction, String dltDestination,
DltSenderContext dltSenderContext, 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(dltSenderContext, "DltSenderContext cannot be null");
this.dltSenderContext = dltSenderContext;
}
public DltAwareProcessor(BiFunction<KIn, VIn, KeyValue<KOut, VOut>> delegateFunction,
BiConsumer<Record<KIn, VIn>, Exception> processorRecordRecoverer) {
this(delegateFunction, System::currentTimeMillis, processorRecordRecoverer);
}
public DltAwareProcessor(BiFunction<KIn, VIn, KeyValue<KOut, VOut>> delegateFunction,
Supplier<Long> recordTimeSupplier, BiConsumer<Record<KIn, VIn>, Exception> processorRecordRecoverer) {
this.delegateFunction = delegateFunction;
this.recordTimeSupplier = recordTimeSupplier;
Assert.notNull(processorRecordRecoverer, "You must provide a valid processor recoverer");
this.processorRecordRecoverer = processorRecordRecoverer;
}
@Override
public void init(ProcessorContext<KOut, VOut> context) {
Processor.super.init(context);
@@ -56,15 +86,14 @@ public class DltAwareProcessor<KIn, VIn, KOut, VOut> implements Processor<KIn, V
public void process(Record<KIn, VIn> record) {
try {
KeyValue<KOut, VOut> keyValue = this.delegateFunction.apply(record.key(), record.value());
//TODO: What should be the timestamp?
Record<KOut, VOut> downstreamRecord = new Record<>(keyValue.key, keyValue.value, System.currentTimeMillis(), record.headers());
Record<KOut, VOut> downstreamRecord = new Record<>(keyValue.key, keyValue.value, recordTimeSupplier.get(), record.headers());
this.context.forward(downstreamRecord);
}
catch (Exception exception) {
StreamBridge streamBridge = this.dltSenderContext.getStreamBridge();
if (streamBridge != null) {
streamBridge.send(dltDestination, record.value());
if (this.processorRecordRecoverer == null) {
this.processorRecordRecoverer = defaultProcessorRecordRecoverer();
}
this.processorRecordRecoverer.accept(record, exception);
}
}
@@ -73,4 +102,13 @@ public class DltAwareProcessor<KIn, VIn, KOut, VOut> implements Processor<KIn, V
Processor.super.close();
}
BiConsumer<Record<KIn, VIn>, Exception> defaultProcessorRecordRecoverer() {
return (r, e) -> {
StreamBridge streamBridge = this.dltSenderContext.getStreamBridge();
if (streamBridge != null) {
streamBridge.send(this.dltDestination, r.value());
}
};
}
}

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

@@ -32,6 +32,7 @@ public class DltSenderContext implements ApplicationContextAware, InitializingBe
private StreamBridge streamBridge;
@Override
public void afterPropertiesSet() {
this.streamBridge = applicationContext.getBean(StreamBridge.class);

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

@@ -392,6 +392,12 @@ public class KafkaStreamsBinderSupportAutoConfiguration {
return new EncodingDecodingBindAdviceHandler();
}
@Bean
@ConditionalOnMissingBean
public DltSenderContext dltSenderContext() {
return new DltSenderContext();
}
@Configuration(proxyBeanMethods = false)
@ConditionalOnMissingBean(value = KafkaStreamsBinderMetrics.class, name = "outerContext")
@ConditionalOnClass(name = "io.micrometer.core.instrument.MeterRegistry")