From a74ea9fe870fe55087610f43a346d8f6cd3dc8e2 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Tue, 17 Oct 2023 13:02:55 -0400 Subject: [PATCH] Introduce a hierarchy for DltAwareProcessor - Common abstraction - RecordRecoverableProcessor which DltAwareProcessor extends --- .../kafka/streams/DltAwareProcessor.java | 63 ++--------- .../kafka/streams/DltPublishingContext.java | 3 +- .../streams/RecordRecoverableProcessor.java | 105 ++++++++++++++++++ 3 files changed, 114 insertions(+), 57 deletions(-) create mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/RecordRecoverableProcessor.java 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 0fcc077c0..3c4c9816c 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 @@ -19,8 +19,6 @@ package org.springframework.cloud.stream.binder.kafka.streams; import java.util.function.BiConsumer; import java.util.function.Function; -import org.apache.kafka.streams.processor.api.Processor; -import org.apache.kafka.streams.processor.api.ProcessorContext; import org.apache.kafka.streams.processor.api.Record; import org.springframework.cloud.stream.function.StreamBridge; @@ -31,39 +29,28 @@ import org.springframework.util.Assert; import org.springframework.util.StringUtils; /** - * Custom {@link Processor} implementation that is capable of sending a record - * to a DLT if the processing fails. + * Custom {@link RecordRecoverableProcessor} that is capable of sending the failed record to a DLT during recovery. * * @author Soby Chacko - * @since 4.1.0 + * @sinc 4.1.0 */ -public class DltAwareProcessor implements Processor { - - /** - * Delegate {@link Function} that is responsible for processing the data. - */ - private final Function, Record> delegateFunction; +public class DltAwareProcessor extends RecordRecoverableProcessor { /** * DLT destination. */ - private String dltDestination; + private final String dltDestination; /** * {@link DltPublishingContext} used for DLT publishing needs. */ - private DltPublishingContext dltPublishingContext; + private final DltPublishingContext dltPublishingContext; /** * A {@link BiConsumer} that does the recovery of a failed record. */ private BiConsumer, Exception> processorRecordRecoverer; - /** - * {@link ProcessorContext} used in the processor. - */ - private ProcessorContext context; - /** * * @param delegateFunction {@link Function} to process the data @@ -72,51 +59,16 @@ public class DltAwareProcessor implements Processor, Record> delegateFunction, String dltDestination, DltPublishingContext dltPublishingContext) { - this.delegateFunction = delegateFunction; + super(delegateFunction); Assert.isTrue(StringUtils.hasText(dltDestination), "DLT Destination topic must be provided."); this.dltDestination = dltDestination; Assert.notNull(dltPublishingContext, "DltSenderContext cannot be null"); this.dltPublishingContext = dltPublishingContext; } - /** - * - * @param delegateFunction {@link Function} to process the data - * @param processorRecordRecoverer {@link BiConsumer} that recovers failed records - */ - public DltAwareProcessor(Function, Record> delegateFunction, - BiConsumer, Exception> processorRecordRecoverer) { - this.delegateFunction = delegateFunction; - Assert.notNull(processorRecordRecoverer, "You must provide a valid processor recoverer"); - this.processorRecordRecoverer = processorRecordRecoverer; - } @Override - public void init(ProcessorContext context) { - Processor.super.init(context); - this.context = context; - } - - @Override - public void process(Record record) { - try { - Record downstreamRecord = this.delegateFunction.apply(record); - this.context.forward(downstreamRecord); - } - catch (Exception exception) { - if (this.processorRecordRecoverer == null) { - this.processorRecordRecoverer = defaultProcessorRecordRecoverer(); - } - this.processorRecordRecoverer.accept(record, exception); - } - } - - @Override - public void close() { - Processor.super.close(); - } - - BiConsumer, Exception> defaultProcessorRecordRecoverer() { + protected BiConsumer, Exception> defaultProcessorRecordRecoverer() { return (r, e) -> { StreamBridge streamBridge = this.dltPublishingContext.getStreamBridge(); if (streamBridge != null) { @@ -126,5 +78,4 @@ public class DltAwareProcessor implements Processor implements Processor { + + private static final Log LOG = LogFactory.getLog(RecordRecoverableProcessor.class); + + /** + * Delegate {@link Function} that is responsible for processing the data. + */ + private final Function, Record> delegateFunction; + + /** + * A {@link BiConsumer} that does the recovery of a failed record. + */ + private BiConsumer, Exception> processorRecordRecoverer; + + /** + * {@link ProcessorContext} used in the processor. + */ + private ProcessorContext context; + + /** + * @param delegateFunction {@link Function} to process the data + */ + public RecordRecoverableProcessor(Function, Record> delegateFunction) { + this.delegateFunction = delegateFunction; + } + + /** + * + * @param delegateFunction {@link Function} to process the data + * @param processorRecordRecoverer {@link BiConsumer} that recovers failed records + */ + public RecordRecoverableProcessor(Function, Record> delegateFunction, + BiConsumer, Exception> processorRecordRecoverer) { + this.delegateFunction = delegateFunction; + Assert.notNull(processorRecordRecoverer, "You must provide a valid processor recoverer"); + this.processorRecordRecoverer = processorRecordRecoverer; + } + + @Override + public void init(ProcessorContext context) { + Processor.super.init(context); + this.context = context; + } + + @Override + public void process(Record record) { + try { + Record downstreamRecord = this.delegateFunction.apply(record); + this.context.forward(downstreamRecord); + } + catch (Exception exception) { + if (this.processorRecordRecoverer == null) { + this.processorRecordRecoverer = defaultProcessorRecordRecoverer(); + } + this.processorRecordRecoverer.accept(record, exception); + } + } + + @Override + public void close() { + Processor.super.close(); + } + + protected BiConsumer, Exception> defaultProcessorRecordRecoverer() { + return (r, e) -> { + RecordRecoverableProcessor.LOG.warn("Runtime Exceptions: ", e); + }; + } + +}