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

Add DltAwareProcessor trace logging

Browse Source 
Since exception is not propogated, adding a trace log allows an optional way for a developer to utilize the defaultRecoverer and still do some basic review of a given exception.
This commit is contained in:
Steven PG
2023-10-18 19:10:04 -04:00
committed by Soby Chacko
parent 0133a95ebd
commit 4a4ce6975a
1 changed files with 6 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -19,6 +19,8 @@ package org.springframework.cloud.stream.binder.kafka.streams;
import java.util.function.BiConsumer;
import java.util.function.Function;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.kafka.streams.processor.api.Record;
import org.springframework.cloud.stream.function.StreamBridge;
@@ -32,10 +34,13 @@ import org.springframework.util.StringUtils;
* Custom {@link RecordRecoverableProcessor} that is capable of sending the failed record to a DLT during recovery.
*
* @author Soby Chacko
* @author Steven Gantz
* @sinc 4.1.0
*/
public class DltAwareProcessor<KIn, VIn, KOut, VOut> extends RecordRecoverableProcessor<KIn, VIn, KOut, VOut> {
private static final Log LOG = LogFactory.getLog(DltAwareProcessor.class);
/**
* DLT destination.
*/
@@ -74,6 +79,7 @@ public class DltAwareProcessor<KIn, VIn, KOut, VOut> extends RecordRecoverablePr
if (streamBridge != null) {
Message<VIn> message = MessageBuilder.withPayload(r.value())
.setHeader(KafkaHeaders.KEY, r.key()).build();
DltAwareProcessor.LOG.trace("Recovered from Exception: ", e);
streamBridge.send(this.dltDestination, message);
}
};