getSerdesForValue())
+ .withName(repartitionName));
+ return repartitioned.process(() -> new DedupKeyValueProcessor<>(storeName, windowDuration),
+ storeName);
}
/**
- * Deduplicate the input stream by applying the deduplicationKeyExtractor function on each record to generate a unique signature for the record.
+ * Deduplicate the input stream by applying the deduplicationKeyExtractor function on each record
+ * to generate a unique signature for the record.
* Uses a window store for the given period of time.
* The input stream should have a String key.
- * ⚠ This constructor should not be used if using the deduplicator multiple times in the same topology. Use {@link DeduplicationUtils#deduplicateWithPredicate(StreamsBuilder, KStream, String storeName, String repartitionName, Duration, Function)} in this scenario.
+ * ⚠ This constructor should not be used if using the deduplicator multiple times in the same topology.
+ * Use {@link
+ * DeduplicationUtils#deduplicateWithPredicate(StreamsBuilder, KStream, String storeName,
+ * String repartitionName, Duration, Function)}
+ * in this scenario.
*
* @param streamsBuilder Stream builder instance for topology editing
* @param initialStream Stream containing the events that should be deduplicated
* @param windowDuration Window of time to keep in the window store
- * @param deduplicationKeyExtractor Function that should extract a deduplication key in String format. This key acts like a comparison vector. A recommended approach is to concatenate all necessary fields in String format to provide a unique identifier for comparison between records.
- * @param Generic Type of the Stream value. Key type is not implemented because using anything other than a String as the key is retarded. You can quote me on this.
+ * @param deduplicationKeyExtractor Function that should extract a deduplication key in String format.
+ * This key acts like a comparison vector.
+ * A recommended approach is to concatenate all necessary fields in String format
+ * to provide a unique identifier for comparison between records.
+ * @param Generic Type of the Stream value.
+ * Key type is not implemented because using anything other
+ * than a String as the key is retarded.
+ * You can quote me on this.
* @return Resulting de-duplicated Stream
*/
- public static KStream> deduplicateWithPredicate(StreamsBuilder streamsBuilder, KStream initialStream, Duration windowDuration, Function deduplicationKeyExtractor) {
- return deduplicateWithPredicate(streamsBuilder, initialStream, DEFAULT_DEDUP_NAME + DEFAULT_WINDOWSTORE, DEFAULT_DEDUP_NAME + DEFAULT_REPARTITION, windowDuration, deduplicationKeyExtractor);
+ public static KStream> deduplicateWithPredicate(
+ StreamsBuilder streamsBuilder, KStream initialStream, Duration windowDuration,
+ Function deduplicationKeyExtractor) {
+ return deduplicateWithPredicate(streamsBuilder, initialStream,
+ DEFAULT_DEDUP_NAME + DEFAULT_WINDOWSTORE, DEFAULT_DEDUP_NAME + DEFAULT_REPARTITION,
+ windowDuration, deduplicationKeyExtractor);
}
/**
- * Deduplicate the input stream by applying the deduplicationKeyExtractor function on each record to generate a unique signature for the record.
+ * Deduplicate the input stream by applying the deduplicationKeyExtractor function
+ * on each record to generate a unique signature for the record.
* Uses a window store for the given period of time.
* The input stream should have a String key.
*
@@ -132,16 +178,31 @@ public static KStream>
* @param storeName Statestore name
* @param repartitionName Repartition topic name
* @param windowDuration Window of time to keep in the window store
- * @param deduplicationKeyExtractor Function that should extract a deduplication key in String format. This key acts like a comparison vector. A recommended approach is to concatenate all necessary fields in String format to provide a unique identifier for comparison between records.
- * @param Generic Type of the Stream value. Key type is not implemented because using anything other than a String as the key is retarded. You can quote me on this.
+ * @param deduplicationKeyExtractor Function that should extract a deduplication key in String format.
+ * This key acts like a comparison vector.
+ * A recommended approach is to concatenate all necessary fields
+ * in String format to provide a unique identifier for comparison between records.
+ * @param Generic Type of the Stream value.
+ * Key type is not implemented because using anything other than
+ * a String as the key is retarded.
+ * You can quote me on this.
* @return Resulting de-duplicated Stream
*/
- public static KStream> deduplicateWithPredicate(StreamsBuilder streamsBuilder, KStream initialStream, String storeName, String repartitionName, Duration windowDuration, Function deduplicationKeyExtractor) {
- StoreBuilder> dedupStore = Stores.timestampedKeyValueStoreBuilder(
- Stores.persistentTimestampedKeyValueStore(storeName), Serdes.String(), SerdesUtils.getSerdesForValue());
+ public static KStream> deduplicateWithPredicate(
+ StreamsBuilder streamsBuilder, KStream initialStream, String storeName,
+ String repartitionName, Duration windowDuration,
+ Function deduplicationKeyExtractor) {
+ StoreBuilder> dedupStore =
+ Stores.timestampedKeyValueStoreBuilder(
+ Stores.persistentTimestampedKeyValueStore(storeName), Serdes.String(),
+ SerdesUtils.getSerdesForValue());
streamsBuilder.addStateStore(dedupStore);
- var repartitioned = initialStream.repartition(Repartitioned.with(Serdes.String(), SerdesUtils.getSerdesForValue()).withName(repartitionName));
- return repartitioned.process(() -> new DedupWithPredicateProcessor<>(storeName, windowDuration, deduplicationKeyExtractor), storeName);
+ var repartitioned = initialStream.repartition(
+ Repartitioned.with(Serdes.String(), SerdesUtils.getSerdesForValue())
+ .withName(repartitionName));
+ return repartitioned.process(
+ () -> new DedupWithPredicateProcessor<>(storeName, windowDuration,
+ deduplicationKeyExtractor), storeName);
}
}
diff --git a/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/DlqDeserializationExceptionHandler.java b/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/DlqDeserializationExceptionHandler.java
index 3c1cc745..75abc4e8 100644
--- a/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/DlqDeserializationExceptionHandler.java
+++ b/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/DlqDeserializationExceptionHandler.java
@@ -2,25 +2,25 @@
import com.michelin.kstreamplify.avro.KafkaError;
import com.michelin.kstreamplify.context.KafkaStreamsExecutionContext;
+import java.util.Map;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.KafkaException;
import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
import org.apache.kafka.streams.processor.ProcessorContext;
-import org.apache.kafka.common.KafkaException;
-
-import java.util.Map;
/**
- * The class managing deserialization exceptions
+ * The class managing deserialization exceptions.
*/
@Slf4j
-public class DlqDeserializationExceptionHandler extends DlqExceptionHandler implements DeserializationExceptionHandler {
+public class DlqDeserializationExceptionHandler extends DlqExceptionHandler
+ implements DeserializationExceptionHandler {
private static final Object GUARD = new Object();
/**
- * Manage deserialization exceptions
+ * Manage deserialization exceptions.
*
* @param processorContext the processor context
* @param consumerRecord the record to deserialize
@@ -28,34 +28,50 @@ public class DlqDeserializationExceptionHandler extends DlqExceptionHandler impl
* @return FAIL or CONTINUE
*/
@Override
- public DeserializationHandlerResponse handle(ProcessorContext processorContext, ConsumerRecord consumerRecord, Exception consumptionException) {
+ public DeserializationHandlerResponse handle(ProcessorContext processorContext,
+ ConsumerRecord consumerRecord,
+ Exception consumptionException) {
if (StringUtils.isBlank(KafkaStreamsExecutionContext.getDlqTopicName())) {
- log.warn("Failed to route deserialization error to the designated DLQ (Dead Letter Queue) topic. Please make sure to define a DLQ topic in your KafkaStreamsStarter bean configuration.");
+ log.warn(
+ "Failed to route deserialization error to the designated DLQ (Dead Letter Queue) topic. "
+ +
+ "Please make sure to define a DLQ topic in your KafkaStreamsStarter bean configuration.");
return DeserializationHandlerResponse.FAIL;
}
try {
var builder = KafkaError.newBuilder();
- enrichWithException(builder, consumptionException, consumerRecord.key(), consumerRecord.value())
- .setContextMessage("An exception occurred during the stream internal deserialization")
- .setOffset(consumerRecord.offset())
- .setPartition(consumerRecord.partition())
- .setTopic(consumerRecord.topic());
+ enrichWithException(builder, consumptionException, consumerRecord.key(),
+ consumerRecord.value())
+ .setContextMessage(
+ "An exception occurred during the stream internal deserialization")
+ .setOffset(consumerRecord.offset())
+ .setPartition(consumerRecord.partition())
+ .setTopic(consumerRecord.topic());
boolean isCausedByKafka = consumptionException.getCause() instanceof KafkaException;
- //If the cause of this exception is a KafkaException and if getCause == sourceException (see Throwable.getCause - including SerializationException)
- //use to handle poison pill => sent message into dlq and continue our life.
- if(isCausedByKafka || consumptionException.getCause() == null) {
- producer.send(new ProducerRecord<>(KafkaStreamsExecutionContext.getDlqTopicName(), consumerRecord.key(), builder.build())).get();
+ // If the cause of this exception is a KafkaException and if getCause == sourceException
+ // (see Throwable.getCause - including SerializationException)
+ // use to handle poison pill => sent message into dlq and continue our life.
+ if (isCausedByKafka || consumptionException.getCause() == null) {
+ producer.send(new ProducerRecord<>(KafkaStreamsExecutionContext.getDlqTopicName(),
+ consumerRecord.key(), builder.build())).get();
return DeserializationHandlerResponse.CONTINUE;
}
} catch (InterruptedException ie) {
- log.error("Interruption while sending the deserialization exception {} for key {}, value {} and topic {} to DLQ topic {}", consumptionException,
- consumerRecord.key(), consumerRecord.value(), consumerRecord.topic(), KafkaStreamsExecutionContext.getDlqTopicName(), ie);
+ log.error(
+ "Interruption while sending the deserialization exception {} for key {}, "
+ + "value {} and topic {} to DLQ topic {}",
+ consumptionException,
+ consumerRecord.key(), consumerRecord.value(), consumerRecord.topic(),
+ KafkaStreamsExecutionContext.getDlqTopicName(), ie);
Thread.currentThread().interrupt();
} catch (Exception e) {
- log.error("Cannot send the deserialization exception {} for key {}, value {} and topic {} to DLQ topic {}", consumptionException,
- consumerRecord.key(), consumerRecord.value(), consumerRecord.topic(), KafkaStreamsExecutionContext.getDlqTopicName(), e);
+ log.error(
+ "Cannot send the deserialization exception {} for key {}, value {} and topic {} to DLQ topic {}",
+ consumptionException,
+ consumerRecord.key(), consumerRecord.value(), consumerRecord.topic(),
+ KafkaStreamsExecutionContext.getDlqTopicName(), e);
}
// here we only have exception like UnknownHostException for example or TimeoutException ...
diff --git a/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/DlqExceptionHandler.java b/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/DlqExceptionHandler.java
index 7d635d25..84e723b8 100644
--- a/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/DlqExceptionHandler.java
+++ b/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/DlqExceptionHandler.java
@@ -2,51 +2,56 @@
import com.michelin.kstreamplify.avro.KafkaError;
import io.confluent.kafka.serializers.KafkaAvroSerializer;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.clients.producer.KafkaProducer;
-import org.apache.kafka.clients.producer.ProducerConfig;
-import org.apache.kafka.common.errors.RecordTooLargeException;
-import org.apache.kafka.common.serialization.ByteArraySerializer;
-
import java.io.PrintWriter;
import java.io.StringWriter;
import java.nio.ByteBuffer;
import java.util.Map;
import java.util.Properties;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.errors.RecordTooLargeException;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
/**
- * The class to manage DLQ exception
+ * The class to manage DLQ exception.
*/
@Slf4j
public abstract class DlqExceptionHandler {
/**
- * The DLQ producer
+ * The DLQ producer.
*/
protected static KafkaProducer producer;
/**
- * Create a producer
+ * Create a producer.
+ *
* @param clientId The producer client id
- * @param configs The producer configs
+ * @param configs The producer configs
*/
protected static void instantiateProducer(String clientId, Map configs) {
Properties properties = new Properties();
properties.putAll(configs);
- properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
- properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName());
+ properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
+ ByteArraySerializer.class.getName());
+ properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
+ KafkaAvroSerializer.class.getName());
properties.setProperty(ProducerConfig.CLIENT_ID_CONFIG, clientId);
producer = new KafkaProducer<>(properties);
}
/**
- * enrich with exception
- * @param builder the error builder
+ * Enrich with exception.
+ *
+ * @param builder the error builder
* @param exception the exception to add
- * @param key the record key
- * @param value the record value
+ * @param key the record key
+ * @param value the record value
* @return the error enriched by the exception
*/
- protected KafkaError.Builder enrichWithException(KafkaError.Builder builder, Exception exception, byte[] key, byte[] value) {
+ protected KafkaError.Builder enrichWithException(KafkaError.Builder builder,
+ Exception exception, byte[] key,
+ byte[] value) {
StringWriter sw = new StringWriter();
PrintWriter pw = new PrintWriter(sw);
exception.printStackTrace(pw);
@@ -54,9 +59,11 @@ protected KafkaError.Builder enrichWithException(KafkaError.Builder builder, Exc
boolean tooLarge = exception instanceof RecordTooLargeException;
return builder
- .setCause(exception.getCause() != null ? exception.getCause().getMessage() : "Unknown cause")
- .setValue(tooLarge ? "The record is too large to be set as value (" + value.length + " bytes). The key will be used instead" : null)
- .setStack(sw.toString())
- .setByteValue(tooLarge ? ByteBuffer.wrap(key) : ByteBuffer.wrap(value));
+ .setCause(
+ exception.getCause() != null ? exception.getCause().getMessage() : "Unknown cause")
+ .setValue(tooLarge ? "The record is too large to be set as value (" + value.length
+ + " bytes). The key will be used instead" : null)
+ .setStack(sw.toString())
+ .setByteValue(tooLarge ? ByteBuffer.wrap(key) : ByteBuffer.wrap(value));
}
}
diff --git a/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/DlqProductionExceptionHandler.java b/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/DlqProductionExceptionHandler.java
index ad685047..7e376adc 100644
--- a/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/DlqProductionExceptionHandler.java
+++ b/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/DlqProductionExceptionHandler.java
@@ -2,31 +2,36 @@
import com.michelin.kstreamplify.avro.KafkaError;
import com.michelin.kstreamplify.context.KafkaStreamsExecutionContext;
+import java.util.Map;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.errors.RetriableException;
import org.apache.kafka.streams.errors.ProductionExceptionHandler;
-import java.util.Map;
-
/**
- * The class managing DLQ production exceptions
+ * The class managing DLQ production exceptions.
*/
@Slf4j
-public class DlqProductionExceptionHandler extends DlqExceptionHandler implements ProductionExceptionHandler {
+public class DlqProductionExceptionHandler extends DlqExceptionHandler
+ implements ProductionExceptionHandler {
private static final Object GUARD = new Object();
/**
- * Manage production exceptions
- * @param producerRecord the record to produce
+ * Manage production exceptions.
+ *
+ * @param producerRecord the record to produce
* @param productionException the exception on producing
* @return FAIL or CONTINUE
*/
@Override
- public ProductionExceptionHandlerResponse handle(ProducerRecord producerRecord, Exception productionException) {
+ public ProductionExceptionHandlerResponse handle(ProducerRecord producerRecord,
+ Exception productionException) {
if (StringUtils.isBlank(KafkaStreamsExecutionContext.getDlqTopicName())) {
- log.warn("Failed to route production error to the designated DLQ (Dead Letter Queue) topic. Please make sure to define a DLQ topic in your KafkaStreamsStarter bean configuration.");
+ log.warn(
+ "Failed to route production error to the designated DLQ (Dead Letter Queue) topic. "
+ +
+ "Please make sure to define a DLQ topic in your KafkaStreamsStarter bean configuration.");
return ProductionExceptionHandlerResponse.FAIL;
}
@@ -35,20 +40,31 @@ public ProductionExceptionHandlerResponse handle(ProducerRecord
if (!retryable) {
try {
var builder = KafkaError.newBuilder();
- enrichWithException(builder, productionException, producerRecord.key(), producerRecord.value())
- .setContextMessage("An exception occurred during the stream internal production")
- .setOffset(-1)
- .setPartition(producerRecord.partition() == null ? -1 : producerRecord.partition())
- .setTopic(producerRecord.topic());
+ enrichWithException(builder, productionException, producerRecord.key(),
+ producerRecord.value())
+ .setContextMessage(
+ "An exception occurred during the stream internal production")
+ .setOffset(-1)
+ .setPartition(
+ producerRecord.partition() == null ? -1 : producerRecord.partition())
+ .setTopic(producerRecord.topic());
- producer.send(new ProducerRecord<>(KafkaStreamsExecutionContext.getDlqTopicName(), producerRecord.key(), builder.build())).get();
+ producer.send(new ProducerRecord<>(KafkaStreamsExecutionContext.getDlqTopicName(),
+ producerRecord.key(), builder.build())).get();
} catch (InterruptedException ie) {
- log.error("Interruption while sending the production exception {} for key {}, value {} and topic {} to DLQ topic {}", productionException,
- producerRecord.key(), producerRecord.value(), producerRecord.topic(), KafkaStreamsExecutionContext.getDlqTopicName(), ie);
+ log.error(
+ "Interruption while sending the production exception {} for key {}, value {} "
+ + "and topic {} to DLQ topic {}",
+ productionException,
+ producerRecord.key(), producerRecord.value(), producerRecord.topic(),
+ KafkaStreamsExecutionContext.getDlqTopicName(), ie);
Thread.currentThread().interrupt();
} catch (Exception e) {
- log.error("Cannot send the production exception {} for key {}, value {} and topic {} to DLQ topic {}", productionException,
- producerRecord.key(), producerRecord.value(), producerRecord.topic(), KafkaStreamsExecutionContext.getDlqTopicName(), e);
+ log.error(
+ "Cannot send the production exception {} for key {}, value {} and topic {} to DLQ topic {}",
+ productionException,
+ producerRecord.key(), producerRecord.value(), producerRecord.topic(),
+ KafkaStreamsExecutionContext.getDlqTopicName(), e);
return ProductionExceptionHandlerResponse.CONTINUE;
}
diff --git a/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/GenericErrorProcessor.java b/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/GenericErrorProcessor.java
index 70aeca41..73e60f5f 100644
--- a/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/GenericErrorProcessor.java
+++ b/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/GenericErrorProcessor.java
@@ -1,23 +1,25 @@
package com.michelin.kstreamplify.error;
import com.michelin.kstreamplify.avro.KafkaError;
+import java.io.PrintWriter;
+import java.io.StringWriter;
import org.apache.kafka.streams.processor.api.FixedKeyProcessor;
import org.apache.kafka.streams.processor.api.FixedKeyProcessorContext;
import org.apache.kafka.streams.processor.api.FixedKeyRecord;
import org.apache.kafka.streams.processor.api.RecordMetadata;
-import java.io.PrintWriter;
-import java.io.StringWriter;
-
/**
- * Generic error processor
+ * Generic error processor.
+ *
* @param The type of the failed record
*/
-public class GenericErrorProcessor implements FixedKeyProcessor, KafkaError> {
+public class GenericErrorProcessor
+ implements FixedKeyProcessor, KafkaError> {
private FixedKeyProcessorContext context;
/**
- * init context
+ * Init context.
+ *
* @param context the context to init
*/
@Override
@@ -26,7 +28,8 @@ public void init(FixedKeyProcessorContext context) {
}
/**
- * process the error
+ * Process the error.
+ *
* @param fixedKeyRecord the record to process an error
*/
@Override
@@ -38,14 +41,16 @@ public void process(FixedKeyRecord> fixedKeyRecord) {
RecordMetadata recordMetadata = context.recordMetadata().orElse(null);
KafkaError error = KafkaError.newBuilder()
- .setCause(fixedKeyRecord.value().getException().getMessage())
- .setContextMessage(fixedKeyRecord.value().getContextMessage())
- .setOffset(recordMetadata != null ? recordMetadata.offset() : -1)
- .setPartition(recordMetadata != null ? recordMetadata.partition() : -1)
- .setStack(sw.toString())
- .setTopic(recordMetadata != null && recordMetadata.topic() != null ? recordMetadata.topic() : "Outside topic context")
- .setValue(fixedKeyRecord.value().getKafkaRecord())
- .build();
+ .setCause(fixedKeyRecord.value().getException().getMessage())
+ .setContextMessage(fixedKeyRecord.value().getContextMessage())
+ .setOffset(recordMetadata != null ? recordMetadata.offset() : -1)
+ .setPartition(recordMetadata != null ? recordMetadata.partition() : -1)
+ .setStack(sw.toString())
+ .setTopic(
+ recordMetadata != null && recordMetadata.topic() != null ? recordMetadata.topic() :
+ "Outside topic context")
+ .setValue(fixedKeyRecord.value().getKafkaRecord())
+ .build();
context.forward(fixedKeyRecord.withValue(error));
}
diff --git a/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/ProcessingError.java b/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/ProcessingError.java
index c4992ecd..0547388c 100644
--- a/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/ProcessingError.java
+++ b/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/ProcessingError.java
@@ -5,31 +5,33 @@
import org.apache.avro.generic.GenericRecord;
/**
- * The processing error class
+ * The processing error class.
+ *
* @param The type of the failed record
*/
@Getter
public class ProcessingError {
/**
- * The exception that occurred
+ * The exception that occurred.
*/
private final Exception exception;
/**
- * The failed Kafka record
+ * The failed Kafka record.
*/
private final String kafkaRecord;
/**
- * A context message defined when the error is caught
+ * A context message defined when the error is caught.
*/
private final String contextMessage;
/**
- * Constructor
- * @param exception The exception
+ * Constructor.
+ *
+ * @param exception The exception
* @param contextMessage The context message
- * @param kafkaRecord The failed Kafka record
+ * @param kafkaRecord The failed Kafka record
*/
public ProcessingError(Exception exception, String contextMessage, V kafkaRecord) {
this.exception = exception;
@@ -43,8 +45,9 @@ public ProcessingError(Exception exception, String contextMessage, V kafkaRecord
}
/**
- * Constructor
- * @param exception The exception
+ * Constructor.
+ *
+ * @param exception The exception
* @param kafkaRecord The failed Kafka record
*/
public ProcessingError(Exception exception, V kafkaRecord) {
diff --git a/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/ProcessingResult.java b/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/ProcessingResult.java
index 9454c353..858ccc63 100644
--- a/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/ProcessingResult.java
+++ b/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/ProcessingResult.java
@@ -4,7 +4,7 @@
import org.apache.kafka.streams.processor.api.Record;
/**
- * The processing result class
+ * The processing result class.
*
* @param The type of the successful record
* @param The type of the failed record
@@ -12,17 +12,17 @@
@Getter
public class ProcessingResult {
/**
- * The successful record
+ * The successful record.
*/
private V value;
/**
- * The failed record wrapped in a processing error
+ * The failed record wrapped in a processing error.
*/
private ProcessingError error;
/**
- * Private constructor that sets the success value
+ * Private constructor that sets the success value.
*
* @param value The success value
*/
@@ -31,7 +31,7 @@ private ProcessingResult(V value) {
}
/**
- * Private constructor that sets the error value
+ * Private constructor that sets the error value.
*
* @param error the ProcessingError containing the
*/
@@ -40,7 +40,7 @@ private ProcessingResult(ProcessingError error) {
}
/**
- * Create a successful processing result
+ * Create a successful processing result.
*
* @param value The successful record value
* @param The type of the successful record
@@ -54,16 +54,19 @@ public static ProcessingResult success(V value) {
/**
* Wraps a record's value with ProcessingResult.success(V value).
- * The resulting stream needs to be handled with TopologyErrorHandler#catchErrors(KStream) for automatic DLQ redirection on failed records.
+ * The resulting stream needs to be handled with TopologyErrorHandler#catchErrors(KStream)
+ * for automatic DLQ redirection on failed records.
*
- * @param record The resulting successful Record from the processor that needs to be wrapped in a ProcessingResult
- * @param The type of the record key
- * @param The type of the ProcessingResult successful value
- * @param The type of the ProcessingResult error value
+ * @param message The resulting successful Record from the processor that needs to be wrapped in a ProcessingResult
+ * @param The type of the record key
+ * @param The type of the ProcessingResult successful value
+ * @param The type of the ProcessingResult error value
* @return The initial Record, with value wrapped in a ProcessingResult
*/
- public static Record> wrapRecordSuccess(Record record) {
- return new Record<>(record.key(), ProcessingResult.success(record.value()), record.timestamp());
+ public static Record> wrapRecordSuccess(
+ Record message) {
+ return new Record<>(message.key(), ProcessingResult.success(message.value()),
+ message.timestamp());
}
/**
@@ -71,7 +74,8 @@ public static Record> wrapRecordSuccess(Re
* Wraps a key, value and timestamp in a Record with ProcessingResult#success(V value) as value.
*
*
- * The resulting stream needs to be handled with TopologyErrorHandler#catchErrors(KStream) for automatic DLQ redirection on failed records.
+ * The resulting stream needs to be handled with TopologyErrorHandler#catchErrors(KStream)
+ * for automatic DLQ redirection on failed records.
*
*
* @param key The key to put in the resulting record
@@ -82,16 +86,18 @@ public static Record> wrapRecordSuccess(Re
* @param The type of the ProcessingResult error value
* @return A Record with value wrapped in a {@link ProcessingResult}
*/
- public static Record> wrapRecordSuccess(K key, V value, long timestamp) {
+ public static Record> wrapRecordSuccess(K key, V value,
+ long timestamp) {
return new Record<>(key, ProcessingResult.success(value), timestamp);
}
/**
*
- * Create a failed processing result
+ * Create a failed processing result.
*
*
- * If you are using this in a Processor, refer to {@link ProcessingResult#wrapRecordFailure(Exception, Record)} for easier syntax.
+ * If you are using this in a Processor, refer to
+ * {@link ProcessingResult#wrapRecordFailure(Exception, Record)} for easier syntax.
*
*
* @param e The exception
@@ -109,7 +115,9 @@ public static ProcessingResult fail(Exception e, V2 failedRecordV
* Create a failed processing result.
*
*
- * If you are using this in a Processor, refer to {@link ProcessingResult#wrapRecordFailure(Exception, Record, String)} for easier syntax.
+ * If you are using this in a Processor, refer to
+ * {@link ProcessingResult#wrapRecordFailure(Exception, Record, String)}
+ * for easier syntax.
*
*
* @param e The exception
@@ -119,7 +127,8 @@ public static ProcessingResult fail(Exception e, V2 failedRecordV
* @param The type of the failed record
* @return A processing result containing the failed record
*/
- public static ProcessingResult fail(Exception e, V2 failedRecordValue, String contextMessage) {
+ public static ProcessingResult fail(Exception e, V2 failedRecordValue,
+ String contextMessage) {
return new ProcessingResult<>(new ProcessingError<>(e, contextMessage, failedRecordValue));
}
@@ -128,18 +137,22 @@ public static ProcessingResult fail(Exception e, V2 failedRecordV
* Wraps a record's value with {@link ProcessingResult#fail(Exception, Object)} )}.
*
*
- * The resulting stream needs to be handled with TopologyErrorHandler#catchErrors(KStream) for automatic DLQ redirection on failed records.
+ * The resulting stream needs to be handled with TopologyErrorHandler#catchErrors(KStream) for automatic
+ * DLQ redirection on failed records.
*
*
* @param e The initial exception
- * @param failedRecord The resulting failed Record from the processor that needs to be wrapped in a {@link ProcessingResult}
+ * @param failedRecord The resulting failed Record from
+ * the processor that needs to be wrapped in a {@link ProcessingResult}
* @param The type of the record key
* @param The type of the ProcessingResult successful value
* @param The type of the ProcessingResult error value
* @return The initial Record, with value wrapped in a {@link ProcessingError} and {@link ProcessingResult}
*/
- public static Record> wrapRecordFailure(Exception e, Record failedRecord) {
- return new Record<>(failedRecord.key(), ProcessingResult.fail(e, failedRecord.value()), failedRecord.timestamp());
+ public static Record> wrapRecordFailure(Exception e,
+ Record failedRecord) {
+ return new Record<>(failedRecord.key(), ProcessingResult.fail(e, failedRecord.value()),
+ failedRecord.timestamp());
}
/**
@@ -147,49 +160,64 @@ public static Record> wrapRecordFailure(Ex
* Wraps a record's value with {@link ProcessingResult#fail(Exception, Object, String)}.
*
*
- * The resulting stream needs to be handled with TopologyErrorHandler#catchErrors(KStream) for automatic DLQ redirection on failed records.
+ * The resulting stream needs to be handled with TopologyErrorHandler#catchErrors(KStream)
+ * for automatic DLQ redirection on failed records.
*
*
* @param e The initial exception
- * @param failedRecord The resulting failed Record from the processor that needs to be wrapped in a {@link ProcessingResult}
+ * @param failedRecord The resulting failed Record from
+ * the processor that needs to be wrapped in a {@link ProcessingResult}
* @param contextMessage The custom context message that will be added in the stack trace
* @param The type of the record key
* @param The type of the ProcessingResult successful value
* @param The type of the ProcessingResult error value
* @return The initial Record, with value wrapped in a {@link ProcessingError} and {@link ProcessingResult}
*/
- public static Record> wrapRecordFailure(Exception e, Record failedRecord, String contextMessage) {
- return new Record<>(failedRecord.key(), ProcessingResult.fail(e, failedRecord.value(), contextMessage), failedRecord.timestamp());
+ public static Record> wrapRecordFailure(Exception e,
+ Record failedRecord,
+ String contextMessage) {
+ return new Record<>(failedRecord.key(),
+ ProcessingResult.fail(e, failedRecord.value(), contextMessage),
+ failedRecord.timestamp());
}
/**
- * Wraps a key, value and timestamp in a Record with {@link ProcessingResult#fail(Exception, Object, String)} as value.
- * The resulting stream needs to be handled with TopologyErrorHandler#catchErrors(KStream) for automatic DLQ redirection on failed records.
+ * Wraps a key, value and timestamp in a Record with {@link ProcessingResult#fail(Exception, Object, String)}
+ * as value.
+ * The resulting stream needs to be handled with TopologyErrorHandler#catchErrors(KStream) for automatic
+ * DLQ redirection on failed records.
*
* @param e The initial exception
* @param key The key to put in the resulting record
- * @param failedValue The resulting failed value from the processor that needs to be wrapped in a {@link ProcessingResult}
+ * @param failedValue The resulting failed value from
+ * the processor that needs to be wrapped in a {@link ProcessingResult}
* @param timestamp The timestamp to apply on the resulting record
* @param The type of the record key
* @param The type of the ProcessingResult successful value
* @param The type of the ProcessingResult error value
* @return A Record with value wrapped in a {@link ProcessingError} and {@link ProcessingResult}
*/
- public static Record> wrapRecordFailure(Exception e, K key, V2 failedValue, long timestamp) {
+ public static Record> wrapRecordFailure(Exception e,
+ K key,
+ V2 failedValue,
+ long timestamp) {
return new Record<>(key, ProcessingResult.fail(e, failedValue), timestamp);
}
/**
*
- * Wraps a key, value and timestamp in a Record with {@link ProcessingResult#fail(Exception, Object, String)} as value.
+ * Wraps a key, value and timestamp in a Record
+ * with {@link ProcessingResult#fail(Exception, Object, String)} as value.
*
*
- * The resulting stream needs to be handled with TopologyErrorHandler#catchErrors(KStream) for automatic DLQ redirection on failed records.
+ * The resulting stream needs to be handled with TopologyErrorHandler#catchErrors(KStream) for automatic
+ * DLQ redirection on failed records.
*
*
* @param e The initial exception
* @param key The key to put in the resulting record
- * @param failedValue The resulting failed value from the processor that needs to be wrapped in a {@link ProcessingResult}
+ * @param failedValue The resulting failed value from the processor
+ * that needs to be wrapped in a {@link ProcessingResult}
* @param timestamp The timestamp to apply on the resulting record
* @param contextMessage The custom context message that will be added in the stack trace
* @param The type of the record key
@@ -197,12 +225,16 @@ public static Record> wrapRecordFailure(Ex
* @param The type of the ProcessingResult error value
* @return A Record with value wrapped in a {@link ProcessingError} and {@link ProcessingResult}
*/
- public static Record> wrapRecordFailure(Exception e, K key, V2 failedValue, long timestamp, String contextMessage) {
+ public static Record> wrapRecordFailure(Exception e,
+ K key,
+ V2 failedValue,
+ long timestamp,
+ String contextMessage) {
return new Record<>(key, ProcessingResult.fail(e, failedValue, contextMessage), timestamp);
}
/**
- * Is the processing result valid ?
+ * Is the processing result valid.
* Is it valid either if it contains a successful value or an error
*
* @return true if valid, false otherwise
diff --git a/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/TopologyErrorHandler.java b/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/TopologyErrorHandler.java
index b3f12ea0..e8e6ee06 100644
--- a/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/TopologyErrorHandler.java
+++ b/kstreamplify-core/src/main/java/com/michelin/kstreamplify/error/TopologyErrorHandler.java
@@ -2,6 +2,7 @@
import com.michelin.kstreamplify.context.KafkaStreamsExecutionContext;
import com.michelin.kstreamplify.utils.SerdesUtils;
+import java.util.Map;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.common.serialization.Serdes;
@@ -11,79 +12,86 @@
import org.apache.kafka.streams.kstream.Named;
import org.apache.kafka.streams.kstream.Produced;
-import java.util.Map;
-
/**
- * The topology error handler class
+ * The topology error handler class.
*/
@Slf4j
public class TopologyErrorHandler {
private static final String BRANCHING_NAME_NOMINAL = "branch-nominal";
- private TopologyErrorHandler() { }
+ private TopologyErrorHandler() {
+ }
/**
- * Catch the errors from the given stream
+ * Catch the errors from the given stream.
+ *
* @param stream The stream of processing result that may contain processing errors
+ * @param The key type
+ * @param The type of the successful record
+ * @param The type of the failed record
* @return A stream filtered from all processing errors
- * @param The key type
- * @param The type of the successful record
- * @param The type of the failed record
*/
- public static KStream catchErrors(KStream> stream) {
+ public static KStream catchErrors(KStream> stream) {
return catchErrors(stream, false);
}
/**
- * Catch the errors from the given stream
- * @param stream The stream of processing result that may contain processing errors
+ * Catch the errors from the given stream.
+ *
+ * @param stream The stream of processing result that may contain processing errors
* @param allowTombstone Allow sending tombstone in DLQ topic or to be returned
+ * @param