This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 7c539f90298 KAFKA-16448: Unify error-callback exception handling
(#16745)
7c539f90298 is described below
commit 7c539f902983a76f14d0cd993e7d6dcbfdacd909
Author: Matthias J. Sax <matth...@confluent.io>
AuthorDate: Sat Aug 3 12:40:51 2024 -0700
KAFKA-16448: Unify error-callback exception handling (#16745)
Follow up code cleanup for KIP-1033.
This PR unifies the handling of both error cases for exception handlers:
- handler throws an exception
- handler returns null
The unification happens for all 5 handler cases:
- deserialzation
- production / serialization
- production / send
- processing
- punctuation
Reviewers: Sebastien Viale <sebastien.vi...@michelin.com>, Loic Greffier
<loic.greff...@michelin.com>, Bill Bejeck <b...@confluent.io>
---
.../errors/DeserializationExceptionHandler.java | 2 +-
.../streams/errors/ProductionExceptionHandler.java | 2 +-
.../internals/DefaultErrorHandlerContext.java | 12 ++
.../internals/FailedProcessingException.java | 7 +-
.../internals/GlobalStateManagerImpl.java | 5 +-
.../streams/processor/internals/ProcessorNode.java | 21 ++-
.../processor/internals/RecordCollectorImpl.java | 106 +++++++----
.../processor/internals/RecordDeserializer.java | 50 ++---
.../streams/processor/internals/StreamTask.java | 49 +++--
.../streams/integration/EosIntegrationTest.java | 5 +-
.../integration/EosV2UpgradeIntegrationTest.java | 6 +-
.../ProcessingExceptionHandlerIntegrationTest.java | 30 +--
.../processor/internals/RecordCollectorTest.java | 194 +++++++++++++++----
.../internals/RecordDeserializerTest.java | 210 ++++++++++++++-------
.../processor/internals/StreamTaskTest.java | 140 ++++++++++----
15 files changed, 589 insertions(+), 250 deletions(-)
diff --git
a/streams/src/main/java/org/apache/kafka/streams/errors/DeserializationExceptionHandler.java
b/streams/src/main/java/org/apache/kafka/streams/errors/DeserializationExceptionHandler.java
index 0d64611de67..198a97cce44 100644
---
a/streams/src/main/java/org/apache/kafka/streams/errors/DeserializationExceptionHandler.java
+++
b/streams/src/main/java/org/apache/kafka/streams/errors/DeserializationExceptionHandler.java
@@ -37,7 +37,7 @@ public interface DeserializationExceptionHandler extends
Configurable {
* @param context processor context
* @param record record that failed deserialization
* @param exception the actual exception
- * @deprecated Since 3.9. Use Please {@link #handle(ErrorHandlerContext,
ConsumerRecord, Exception)}
+ * @deprecated Since 3.9. Use {@link #handle(ErrorHandlerContext,
ConsumerRecord, Exception)} instead.
*/
@Deprecated
default DeserializationHandlerResponse handle(final ProcessorContext
context,
diff --git
a/streams/src/main/java/org/apache/kafka/streams/errors/ProductionExceptionHandler.java
b/streams/src/main/java/org/apache/kafka/streams/errors/ProductionExceptionHandler.java
index 25aa00f7a92..939b1ecbcd6 100644
---
a/streams/src/main/java/org/apache/kafka/streams/errors/ProductionExceptionHandler.java
+++
b/streams/src/main/java/org/apache/kafka/streams/errors/ProductionExceptionHandler.java
@@ -59,7 +59,7 @@ public interface ProductionExceptionHandler extends
Configurable {
*
* @param record the record that failed to serialize
* @param exception the exception that occurred during serialization
- * @deprecated Since 3.9. Use {@link #handle(ErrorHandlerContext,
ProducerRecord, Exception)} instead.
+ * @deprecated Since 3.9. Use {@link
#handleSerializationException(ErrorHandlerContext, ProducerRecord, Exception,
SerializationExceptionOrigin)} instead.
*/
@Deprecated
default ProductionExceptionHandlerResponse
handleSerializationException(final ProducerRecord record,
diff --git
a/streams/src/main/java/org/apache/kafka/streams/errors/internals/DefaultErrorHandlerContext.java
b/streams/src/main/java/org/apache/kafka/streams/errors/internals/DefaultErrorHandlerContext.java
index c907ff3eb89..77500ce3c36 100644
---
a/streams/src/main/java/org/apache/kafka/streams/errors/internals/DefaultErrorHandlerContext.java
+++
b/streams/src/main/java/org/apache/kafka/streams/errors/internals/DefaultErrorHandlerContext.java
@@ -81,6 +81,18 @@ public class DefaultErrorHandlerContext implements
ErrorHandlerContext {
return taskId;
}
+ @Override
+ public String toString() {
+ // we do exclude headers on purpose, to not accidentally log user data
+ return "ErrorHandlerContext{" +
+ "topic='" + topic + '\'' +
+ ", partition=" + partition +
+ ", offset=" + offset +
+ ", processorNodeId='" + processorNodeId + '\'' +
+ ", taskId=" + taskId +
+ '}';
+ }
+
public Optional<ProcessorContext> processorContext() {
return Optional.ofNullable(processorContext);
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/errors/internals/FailedProcessingException.java
b/streams/src/main/java/org/apache/kafka/streams/errors/internals/FailedProcessingException.java
index 25f2ae9f6cc..03d687d2687 100644
---
a/streams/src/main/java/org/apache/kafka/streams/errors/internals/FailedProcessingException.java
+++
b/streams/src/main/java/org/apache/kafka/streams/errors/internals/FailedProcessingException.java
@@ -25,7 +25,12 @@ import org.apache.kafka.streams.errors.StreamsException;
public class FailedProcessingException extends StreamsException {
private static final long serialVersionUID = 1L;
+ public FailedProcessingException(final String errorMessage, final
Exception exception) {
+ super(errorMessage, exception);
+ }
+
public FailedProcessingException(final Exception exception) {
- super(exception);
+ // we need to explicitly set `message` to `null` here
+ super(null, exception);
}
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
index 6b7214a9ed1..12d4c6c603d 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
@@ -319,7 +319,7 @@ public class GlobalStateManagerImpl implements
GlobalStateManager {
record.headers()));
restoreCount++;
}
- } catch (final Exception deserializationException) {
+ } catch (final RuntimeException deserializationException) {
handleDeserializationFailure(
deserializationExceptionHandler,
globalProcessorContext,
@@ -330,7 +330,8 @@ public class GlobalStateManagerImpl implements
GlobalStateManager {
Thread.currentThread().getName(),
globalProcessorContext.taskId().toString(),
globalProcessorContext.metrics()
- )
+ ),
+ null
);
}
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java
index 3ccfcf24905..07cce0730ee 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java
@@ -38,6 +38,7 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
import static
org.apache.kafka.streams.StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG;
@@ -202,7 +203,7 @@ public class ProcessorNode<KIn, VIn, KOut, VOut> {
} catch (final FailedProcessingException | TaskCorruptedException |
TaskMigratedException e) {
// Rethrow exceptions that should not be handled here
throw e;
- } catch (final RuntimeException e) {
+ } catch (final RuntimeException processingException) {
final ErrorHandlerContext errorHandlerContext = new
DefaultErrorHandlerContext(
null, // only required to pass for
DeserializationExceptionHandler
internalProcessorContext.topic(),
@@ -213,18 +214,26 @@ public class ProcessorNode<KIn, VIn, KOut, VOut> {
internalProcessorContext.taskId());
final ProcessingExceptionHandler.ProcessingHandlerResponse
response;
-
try {
- response =
processingExceptionHandler.handle(errorHandlerContext, record, e);
- } catch (final Exception fatalUserException) {
- throw new FailedProcessingException(fatalUserException);
+ response = Objects.requireNonNull(
+ processingExceptionHandler.handle(errorHandlerContext,
record, processingException),
+ "Invalid ProductionExceptionHandler response."
+ );
+ } catch (final RuntimeException fatalUserException) {
+ log.error(
+ "Processing error callback failed after processing error
for record: {}",
+ errorHandlerContext,
+ processingException
+ );
+ throw new FailedProcessingException("Fatal user code error in
processing error callback", fatalUserException);
}
+
if (response ==
ProcessingExceptionHandler.ProcessingHandlerResponse.FAIL) {
log.error("Processing exception handler is set to fail upon" +
" a processing error. If you would rather have the
streaming pipeline" +
" continue after a processing error, please set the " +
PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG + "
appropriately.");
- throw new FailedProcessingException(e);
+ throw new FailedProcessingException(processingException);
} else {
droppedRecordsSensor.record();
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java
index a4dc0a68062..e471587ed0d 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java
@@ -45,6 +45,7 @@ import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.errors.TaskCorruptedException;
import org.apache.kafka.streams.errors.TaskMigratedException;
import org.apache.kafka.streams.errors.internals.DefaultErrorHandlerContext;
+import org.apache.kafka.streams.errors.internals.FailedProcessingException;
import org.apache.kafka.streams.processor.StreamPartitioner;
import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
@@ -59,6 +60,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
@@ -208,7 +210,7 @@ public class RecordCollectorImpl implements RecordCollector
{
key,
keySerializer,
exception);
- } catch (final Exception exception) {
+ } catch (final RuntimeException serializationException) {
handleException(
ProductionExceptionHandler.SerializationExceptionOrigin.KEY,
topic,
@@ -219,7 +221,7 @@ public class RecordCollectorImpl implements RecordCollector
{
timestamp,
processorNodeId,
context,
- exception);
+ serializationException);
return;
}
@@ -232,7 +234,7 @@ public class RecordCollectorImpl implements RecordCollector
{
value,
valueSerializer,
exception);
- } catch (final Exception exception) {
+ } catch (final RuntimeException serializationException) {
handleException(
ProductionExceptionHandler.SerializationExceptionOrigin.VALUE,
topic,
@@ -243,7 +245,7 @@ public class RecordCollectorImpl implements RecordCollector
{
timestamp,
processorNodeId,
context,
- exception);
+ serializationException);
return;
}
@@ -297,42 +299,51 @@ public class RecordCollectorImpl implements
RecordCollector {
final Long timestamp,
final String processorNodeId,
final InternalProcessorContext<Void,
Void> context,
- final Exception exception) {
+ final RuntimeException
serializationException) {
+ log.debug(String.format("Error serializing record for topic %s",
topic), serializationException);
+
+ final DefaultErrorHandlerContext errorHandlerContext = new
DefaultErrorHandlerContext(
+ null, // only required to pass for DeserializationExceptionHandler
+ context.recordContext().topic(),
+ context.recordContext().partition(),
+ context.recordContext().offset(),
+ context.recordContext().headers(),
+ processorNodeId,
+ taskId
+ );
final ProducerRecord<K, V> record = new ProducerRecord<>(topic,
partition, timestamp, key, value, headers);
- final ProductionExceptionHandlerResponse response;
-
- log.debug(String.format("Error serializing record to topic %s",
topic), exception);
+ final ProductionExceptionHandlerResponse response;
try {
- final DefaultErrorHandlerContext errorHandlerContext = new
DefaultErrorHandlerContext(
- null, // only required to pass for
DeserializationExceptionHandler
- context.recordContext().topic(),
- context.recordContext().partition(),
- context.recordContext().offset(),
- context.recordContext().headers(),
- processorNodeId,
- taskId
+ response = Objects.requireNonNull(
+
productionExceptionHandler.handleSerializationException(errorHandlerContext,
record, serializationException, origin),
+ "Invalid ProductionExceptionHandler response."
);
- response =
productionExceptionHandler.handleSerializationException(errorHandlerContext,
record, exception, origin);
- } catch (final Exception e) {
- log.error("Fatal when handling serialization exception", e);
- recordSendError(topic, e, null, context, processorNodeId);
- return;
+ } catch (final RuntimeException fatalUserException) {
+ log.error(
+ String.format(
+ "Production error callback failed after serialization
error for record %s: %s",
+ origin.toString().toLowerCase(Locale.ROOT),
+ errorHandlerContext
+ ),
+ serializationException
+ );
+ throw new FailedProcessingException("Fatal user code error in
production error callback", fatalUserException);
}
if (response == ProductionExceptionHandlerResponse.FAIL) {
throw new StreamsException(
String.format(
"Unable to serialize record. ProducerRecord(topic=[%s],
partition=[%d], timestamp=[%d]",
- topic,
- partition,
- timestamp),
- exception
+ topic,
+ partition,
+ timestamp),
+ serializationException
);
}
log.warn("Unable to serialize record, continue processing. " +
- "ProducerRecord(topic=[{}], partition=[{}],
timestamp=[{}])",
+ "ProducerRecord(topic=[{}], partition=[{}],
timestamp=[{}])",
topic,
partition,
timestamp);
@@ -364,24 +375,24 @@ public class RecordCollectorImpl implements
RecordCollector {
}
private void recordSendError(final String topic,
- final Exception exception,
+ final Exception productionException,
final ProducerRecord<byte[], byte[]>
serializedRecord,
final InternalProcessorContext<Void, Void>
context,
final String processorNodeId) {
- String errorMessage = String.format(SEND_EXCEPTION_MESSAGE, topic,
taskId, exception.toString());
+ String errorMessage = String.format(SEND_EXCEPTION_MESSAGE, topic,
taskId, productionException.toString());
- if (isFatalException(exception)) {
+ if (isFatalException(productionException)) {
errorMessage += "\nWritten offsets would not be recorded and no
more records would be sent since this is a fatal error.";
- sendException.set(new StreamsException(errorMessage, exception));
- } else if (exception instanceof ProducerFencedException ||
- exception instanceof InvalidPidMappingException ||
- exception instanceof InvalidProducerEpochException ||
- exception instanceof OutOfOrderSequenceException) {
+ sendException.set(new StreamsException(errorMessage,
productionException));
+ } else if (productionException instanceof ProducerFencedException ||
+ productionException instanceof InvalidPidMappingException ||
+ productionException instanceof InvalidProducerEpochException ||
+ productionException instanceof OutOfOrderSequenceException) {
errorMessage += "\nWritten offsets would not be recorded and no
more records would be sent since the producer is fenced, " +
"indicating the task may be migrated out";
- sendException.set(new TaskMigratedException(errorMessage,
exception));
+ sendException.set(new TaskMigratedException(errorMessage,
productionException));
} else {
- if (isRetriable(exception)) {
+ if (isRetriable(productionException)) {
errorMessage += "\nThe broker is either slow or in bad state
(like not having enough replicas) in responding the request, " +
"or the connection to broker was interrupted sending the
request or receiving the response. " +
"\nConsider overwriting `max.block.ms` and /or " +
@@ -398,17 +409,34 @@ public class RecordCollectorImpl implements
RecordCollector {
taskId
);
- if (productionExceptionHandler.handle(errorHandlerContext,
serializedRecord, exception) == ProductionExceptionHandlerResponse.FAIL) {
+ final ProductionExceptionHandlerResponse response;
+ try {
+ response = Objects.requireNonNull(
+ productionExceptionHandler.handle(errorHandlerContext,
serializedRecord, productionException),
+ "Invalid ProductionExceptionHandler response."
+ );
+ } catch (final RuntimeException fatalUserException) {
+ log.error(
+ "Production error callback failed after production
error for record {}",
+ serializedRecord,
+ productionException
+ );
+ sendException.set(new FailedProcessingException("Fatal
user code error in production error callback", fatalUserException));
+ return;
+ }
+
+ if (response == ProductionExceptionHandlerResponse.FAIL) {
errorMessage += "\nException handler choose to FAIL the
processing, no more records would be sent.";
- sendException.set(new StreamsException(errorMessage,
exception));
+ sendException.set(new StreamsException(errorMessage,
productionException));
} else {
errorMessage += "\nException handler choose to CONTINUE
processing in spite of this error but written offsets would not be recorded.";
droppedRecordsSensor.record();
}
+
}
}
- log.error(errorMessage, exception);
+ log.error(errorMessage, productionException);
}
/**
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordDeserializer.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordDeserializer.java
index 8ee2dc014eb..5fc03352ecc 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordDeserializer.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordDeserializer.java
@@ -21,12 +21,14 @@ import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
+import
org.apache.kafka.streams.errors.DeserializationExceptionHandler.DeserializationHandlerResponse;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.errors.internals.DefaultErrorHandlerContext;
import org.apache.kafka.streams.processor.api.ProcessorContext;
import org.slf4j.Logger;
+import java.util.Objects;
import java.util.Optional;
import static
org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG;
@@ -49,7 +51,7 @@ public class RecordDeserializer {
/**
* @throws StreamsException if a deserialization error occurs and the
deserialization callback returns
- * {@link
DeserializationExceptionHandler.DeserializationHandlerResponse#FAIL FAIL}
+ * {@link DeserializationHandlerResponse#FAIL
FAIL}
* or throws an exception itself
*/
ConsumerRecord<Object, Object> deserialize(final ProcessorContext<?, ?>
processorContext,
@@ -69,7 +71,7 @@ public class RecordDeserializer {
rawRecord.headers(),
Optional.empty()
);
- } catch (final Exception deserializationException) {
+ } catch (final RuntimeException deserializationException) {
handleDeserializationFailure(deserializationExceptionHandler,
processorContext, deserializationException, rawRecord, log,
droppedRecordsSensor, sourceNode().name());
return null; // 'handleDeserializationFailure' would either throw
or swallow -- if we swallow we need to skip the record by returning 'null'
}
@@ -77,39 +79,37 @@ public class RecordDeserializer {
public static void handleDeserializationFailure(final
DeserializationExceptionHandler deserializationExceptionHandler,
final ProcessorContext<?,
?> processorContext,
- final Exception
deserializationException,
- final
ConsumerRecord<byte[], byte[]> rawRecord,
- final Logger log,
- final Sensor
droppedRecordsSensor) {
- handleDeserializationFailure(deserializationExceptionHandler,
processorContext, deserializationException, rawRecord, log,
droppedRecordsSensor, null);
- }
-
- public static void handleDeserializationFailure(final
DeserializationExceptionHandler deserializationExceptionHandler,
- final ProcessorContext<?,
?> processorContext,
- final Exception
deserializationException,
+ final RuntimeException
deserializationException,
final
ConsumerRecord<byte[], byte[]> rawRecord,
final Logger log,
final Sensor
droppedRecordsSensor,
final String
sourceNodeName) {
- final DeserializationExceptionHandler.DeserializationHandlerResponse
response;
+
+ final DefaultErrorHandlerContext errorHandlerContext = new
DefaultErrorHandlerContext(
+ (InternalProcessorContext<?, ?>) processorContext,
+ rawRecord.topic(),
+ rawRecord.partition(),
+ rawRecord.offset(),
+ rawRecord.headers(),
+ sourceNodeName,
+ processorContext.taskId());
+
+ final DeserializationHandlerResponse response;
try {
- final DefaultErrorHandlerContext errorHandlerContext = new
DefaultErrorHandlerContext(
- (InternalProcessorContext<?, ?>) processorContext,
- rawRecord.topic(),
- rawRecord.partition(),
- rawRecord.offset(),
- rawRecord.headers(),
- sourceNodeName,
- processorContext.taskId());
- response =
deserializationExceptionHandler.handle(errorHandlerContext, rawRecord,
deserializationException);
- } catch (final Exception fatalUserException) {
+ response = Objects.requireNonNull(
+ deserializationExceptionHandler.handle(errorHandlerContext,
rawRecord, deserializationException),
+ "Invalid DeserializationExceptionHandler response."
+ );
+ } catch (final RuntimeException fatalUserException) {
log.error(
"Deserialization error callback failed after deserialization
error for record {}",
rawRecord,
- deserializationException);
+ deserializationException
+ );
throw new StreamsException("Fatal user code error in
deserialization error callback", fatalUserException);
}
- if (response ==
DeserializationExceptionHandler.DeserializationHandlerResponse.FAIL) {
+
+ if (response == DeserializationHandlerResponse.FAIL) {
throw new StreamsException("Deserialization exception handler is
set to fail upon" +
" a deserialization error. If you would rather have the
streaming pipeline" +
" continue after a deserialization error, please set the " +
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
index 6f2edd442b0..f08cfa7fd67 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
@@ -59,6 +59,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.function.Function;
@@ -807,7 +808,7 @@ public class StreamTask extends AbstractTask implements
ProcessorNodePunctuator,
}
} catch (final FailedProcessingException failedProcessingException) {
// Do not keep the failed processing exception in the stack trace
- handleException(failedProcessingException.getCause());
+ handleException(failedProcessingException.getMessage(),
failedProcessingException.getCause());
} catch (final StreamsException exception) {
record = null;
throw exception;
@@ -820,19 +821,25 @@ public class StreamTask extends AbstractTask implements
ProcessorNodePunctuator,
return true;
}
- private void handleException(final Throwable e) {
- final StreamsException error = new StreamsException(
+ private void handleException(final Throwable originalException) {
+ handleException(
String.format(
- "Exception caught in process. taskId=%s, processor=%s,
topic=%s, partition=%d, offset=%d, stacktrace=%s",
+ "Exception caught in process. taskId=%s, processor=%s,
topic=%s, partition=%d, offset=%d",
id(),
processorContext.currentNode().name(),
record.topic(),
record.partition(),
- record.offset(),
- getStacktraceString(e)
+ record.offset()
),
- e
- );
+ originalException);
+ }
+
+ private void handleException(final String errorMessage, final Throwable
originalException) {
+ if (errorMessage == null) {
+ handleException(originalException);
+ }
+
+ final StreamsException error = new StreamsException(errorMessage,
originalException);
record = null;
throw error;
@@ -920,11 +927,18 @@ public class StreamTask extends AbstractTask implements
ProcessorNodePunctuator,
try {
maybeMeasureLatency(() -> punctuator.punctuate(timestamp), time,
punctuateLatencySensor);
+ } catch (final TimeoutException timeoutException) {
+ if (!eosEnabled) {
+ throw timeoutException;
+ } else {
+ record = null;
+ throw new TaskCorruptedException(Collections.singleton(id));
+ }
} catch (final FailedProcessingException e) {
throw createStreamsException(node.name(), e.getCause());
} catch (final TaskCorruptedException | TaskMigratedException e) {
throw e;
- } catch (final Exception e) {
+ } catch (final RuntimeException processingException) {
final ErrorHandlerContext errorHandlerContext = new
DefaultErrorHandlerContext(
null,
recordContext.topic(),
@@ -936,11 +950,18 @@ public class StreamTask extends AbstractTask implements
ProcessorNodePunctuator,
);
final ProcessingExceptionHandler.ProcessingHandlerResponse
response;
-
try {
- response =
processingExceptionHandler.handle(errorHandlerContext, null, e);
- } catch (final Exception fatalUserException) {
- throw new FailedProcessingException(fatalUserException);
+ response = Objects.requireNonNull(
+ processingExceptionHandler.handle(errorHandlerContext,
null, processingException),
+ "Invalid ProcessingExceptionHandler response."
+ );
+ } catch (final RuntimeException fatalUserException) {
+ log.error(
+ "Processing error callback failed after processing error
for record: {}",
+ errorHandlerContext,
+ processingException
+ );
+ throw new FailedProcessingException("Fatal user code error in
processing error callback", fatalUserException);
}
if (response ==
ProcessingExceptionHandler.ProcessingHandlerResponse.FAIL) {
@@ -949,7 +970,7 @@ public class StreamTask extends AbstractTask implements
ProcessorNodePunctuator,
" continue after a processing error, please set the " +
PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG + "
appropriately.");
- throw createStreamsException(node.name(), e);
+ throw createStreamsException(node.name(), processingException);
} else {
droppedRecordsSensor.record();
}
diff --git
a/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java
b/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java
index 94e48ee3d49..3b588439e92 100644
---
a/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java
@@ -39,6 +39,7 @@ import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.StreamsConfig.InternalConfig;
import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.errors.TaskCorruptedException;
import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
@@ -1170,7 +1171,9 @@ public class EosIntegrationTest {
final KafkaStreams streams = new KafkaStreams(builder.build(), config);
streams.setUncaughtExceptionHandler((t, e) -> {
- if (uncaughtException != null ||
!e.getMessage().contains("Injected test exception")) {
+ if (uncaughtException != null ||
+ !(e instanceof StreamsException) ||
+ !e.getCause().getMessage().equals("Injected test exception."))
{
e.printStackTrace(System.err);
hasUnexpectedError = true;
}
diff --git
a/streams/src/test/java/org/apache/kafka/streams/integration/EosV2UpgradeIntegrationTest.java
b/streams/src/test/java/org/apache/kafka/streams/integration/EosV2UpgradeIntegrationTest.java
index 8653f69fca6..4dd8e5697c4 100644
---
a/streams/src/test/java/org/apache/kafka/streams/integration/EosV2UpgradeIntegrationTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/integration/EosV2UpgradeIntegrationTest.java
@@ -35,6 +35,7 @@ import org.apache.kafka.streams.StoreQueryParameters;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.StreamsConfig.InternalConfig;
+import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler;
import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
@@ -947,8 +948,9 @@ public class EosV2UpgradeIntegrationTest {
} else {
int exceptionCount = exceptionCounts.get(appDir);
// should only have our injected exception or commit
exception, and 2 exceptions for each stream
- if (++exceptionCount > 2 || !(e instanceof RuntimeException) ||
- !(e.getMessage().contains("test exception"))) {
+ if (++exceptionCount > 2 ||
+ !(e instanceof StreamsException) ||
+ !(e.getCause().getMessage().endsWith(" test exception.")))
{
// The exception won't cause the test fail since we
actually "expected" exception thrown and failed the stream.
// So, log to stderr for debugging when the exception is
not what we expected, and fail in the main thread
e.printStackTrace(System.err);
diff --git
a/streams/src/test/java/org/apache/kafka/streams/integration/ProcessingExceptionHandlerIntegrationTest.java
b/streams/src/test/java/org/apache/kafka/streams/integration/ProcessingExceptionHandlerIntegrationTest.java
index 61b5ed16bb1..d0c32310550 100644
---
a/streams/src/test/java/org/apache/kafka/streams/integration/ProcessingExceptionHandlerIntegrationTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/integration/ProcessingExceptionHandlerIntegrationTest.java
@@ -51,6 +51,7 @@ import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertIterableEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -61,7 +62,7 @@ public class ProcessingExceptionHandlerIntegrationTest {
private final String threadId = Thread.currentThread().getName();
@Test
- public void shouldFailWhenProcessingExceptionOccurs() {
+ public void
shouldFailWhenProcessingExceptionOccursIfExceptionHandlerReturnsFail() {
final List<KeyValue<String, String>> events = Arrays.asList(
new KeyValue<>("ID123-1", "ID123-A1"),
new KeyValue<>("ID123-2-ERR", "ID123-A2"),
@@ -93,8 +94,7 @@ public class ProcessingExceptionHandlerIntegrationTest {
assertTrue(exception.getMessage().contains("Exception caught in
process. "
+ "taskId=0_0, processor=KSTREAM-SOURCE-0000000000,
topic=TOPIC_NAME, "
- + "partition=0, offset=1,
stacktrace=java.lang.RuntimeException: "
- + "Exception should be handled by processing exception
handler"));
+ + "partition=0, offset=1"));
assertEquals(1,
processor.theCapturedProcessor().processed().size());
assertIterableEquals(expectedProcessedRecords,
processor.theCapturedProcessor().processed());
@@ -107,7 +107,7 @@ public class ProcessingExceptionHandlerIntegrationTest {
}
@Test
- public void shouldContinueWhenProcessingExceptionOccurs() {
+ public void
shouldContinueWhenProcessingExceptionOccursIfExceptionHandlerReturnsContinue() {
final List<KeyValue<String, String>> events = Arrays.asList(
new KeyValue<>("ID123-1", "ID123-A1"),
new KeyValue<>("ID123-2-ERR", "ID123-A2"),
@@ -182,8 +182,7 @@ public class ProcessingExceptionHandlerIntegrationTest {
final StreamsException e = assertThrows(StreamsException.class, ()
-> inputTopic.pipeInput(eventError.key, eventError.value, Instant.EPOCH));
assertTrue(e.getMessage().contains("Exception caught in process. "
+ "taskId=0_0, processor=KSTREAM-SOURCE-0000000000,
topic=TOPIC_NAME, "
- + "partition=0, offset=1,
stacktrace=java.lang.RuntimeException: "
- + "Exception should be handled by processing exception
handler"));
+ + "partition=0, offset=1"));
assertFalse(isExecuted.get());
}
}
@@ -222,9 +221,9 @@ public class ProcessingExceptionHandlerIntegrationTest {
}
@Test
- public void
shouldStopProcessingWhenFatalUserExceptionInFailProcessingExceptionHandler() {
+ public void
shouldStopProcessingWhenProcessingExceptionHandlerReturnsNull() {
final KeyValue<String, String> event = new KeyValue<>("ID123-1",
"ID123-A1");
- final KeyValue<String, String> eventError = new
KeyValue<>("ID123-ERR-FATAL", "ID123-A2");
+ final KeyValue<String, String> eventError = new
KeyValue<>("ID123-ERR-NULL", "ID123-A2");
final MockProcessorSupplier<String, String, Void, Void> processor =
new MockProcessorSupplier<>();
final StreamsBuilder builder = new StreamsBuilder();
@@ -241,7 +240,7 @@ public class ProcessingExceptionHandlerIntegrationTest {
.process(processor);
final Properties properties = new Properties();
-
properties.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG,
FailProcessingExceptionHandlerMockTest.class);
+
properties.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG,
ContinueProcessingExceptionHandlerMockTest.class);
try (final TopologyTestDriver driver = new
TopologyTestDriver(builder.build(), properties, Instant.ofEpochMilli(0L))) {
final TestInputTopic<String, String> inputTopic =
driver.createInputTopic("TOPIC_NAME", new StringSerializer(), new
StringSerializer());
@@ -250,13 +249,15 @@ public class ProcessingExceptionHandlerIntegrationTest {
assertTrue(isExecuted.get());
isExecuted.set(false);
final StreamsException e = assertThrows(StreamsException.class, ()
-> inputTopic.pipeInput(eventError.key, eventError.value, Instant.EPOCH));
- assertEquals("KABOOM!", e.getCause().getMessage());
+ assertEquals("Fatal user code error in processing error callback",
e.getMessage());
+ assertInstanceOf(NullPointerException.class, e.getCause());
+ assertEquals("Invalid ProductionExceptionHandler response.",
e.getCause().getMessage());
assertFalse(isExecuted.get());
}
}
@Test
- public void
shouldStopProcessingWhenFatalUserExceptionInContinueProcessingExceptionHandler()
{
+ public void
shouldStopProcessingWhenFatalUserExceptionProcessingExceptionHandler() {
final KeyValue<String, String> event = new KeyValue<>("ID123-1",
"ID123-A1");
final KeyValue<String, String> eventError = new
KeyValue<>("ID123-ERR-FATAL", "ID123-A2");
@@ -284,6 +285,7 @@ public class ProcessingExceptionHandlerIntegrationTest {
assertTrue(isExecuted.get());
isExecuted.set(false);
final StreamsException e = assertThrows(StreamsException.class, ()
-> inputTopic.pipeInput(eventError.key, eventError.value, Instant.EPOCH));
+ assertEquals("Fatal user code error in processing error callback",
e.getMessage());
assertEquals("KABOOM!", e.getCause().getMessage());
assertFalse(isExecuted.get());
}
@@ -295,6 +297,9 @@ public class ProcessingExceptionHandlerIntegrationTest {
if (((String) record.key()).contains("FATAL")) {
throw new RuntimeException("KABOOM!");
}
+ if (((String) record.key()).contains("NULL")) {
+ return null;
+ }
assertProcessingExceptionHandlerInputs(context, record, exception);
return
ProcessingExceptionHandler.ProcessingHandlerResponse.CONTINUE;
}
@@ -308,9 +313,6 @@ public class ProcessingExceptionHandlerIntegrationTest {
public static class FailProcessingExceptionHandlerMockTest implements
ProcessingExceptionHandler {
@Override
public ProcessingExceptionHandler.ProcessingHandlerResponse
handle(final ErrorHandlerContext context, final Record<?, ?> record, final
Exception exception) {
- if (((String) record.key()).contains("FATAL")) {
- throw new RuntimeException("KABOOM!");
- }
assertProcessingExceptionHandlerInputs(context, record, exception);
return ProcessingExceptionHandler.ProcessingHandlerResponse.FAIL;
}
diff --git
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java
index 26231028851..ec8f6e5a9f9 100644
---
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java
@@ -51,6 +51,8 @@ import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.errors.DefaultProductionExceptionHandler;
import org.apache.kafka.streams.errors.ErrorHandlerContext;
import org.apache.kafka.streams.errors.ProductionExceptionHandler;
+import
org.apache.kafka.streams.errors.ProductionExceptionHandler.ProductionExceptionHandlerResponse;
+import
org.apache.kafka.streams.errors.ProductionExceptionHandler.SerializationExceptionOrigin;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.errors.TaskMigratedException;
import org.apache.kafka.streams.processor.StreamPartitioner;
@@ -1242,7 +1244,7 @@ public class RecordCollectorTest {
logContext,
taskId,
getExceptionalStreamsProducerOnSend(exception),
- new
ProductionExceptionHandlerMock(ProductionExceptionHandler.ProductionExceptionHandlerResponse.CONTINUE),
+ new
ProductionExceptionHandlerMock(Optional.of(ProductionExceptionHandlerResponse.CONTINUE)),
streamsMetrics,
topology
);
@@ -1269,7 +1271,7 @@ public class RecordCollectorTest {
logContext,
taskId,
getExceptionalStreamsProducerOnSend(exception),
- new
ProductionExceptionHandlerMock(ProductionExceptionHandler.ProductionExceptionHandlerResponse.CONTINUE),
+ new
ProductionExceptionHandlerMock(Optional.of(ProductionExceptionHandlerResponse.CONTINUE)),
streamsMetrics,
topology
);
@@ -1293,7 +1295,7 @@ public class RecordCollectorTest {
logContext,
taskId,
getExceptionalStreamsProducerOnSend(exception),
- new
ProductionExceptionHandlerMock(ProductionExceptionHandler.ProductionExceptionHandlerResponse.CONTINUE),
+ new
ProductionExceptionHandlerMock(Optional.of(ProductionExceptionHandlerResponse.CONTINUE)),
streamsMetrics,
topology
);
@@ -1317,7 +1319,7 @@ public class RecordCollectorTest {
taskId,
getExceptionalStreamsProducerOnSend(new
RuntimeException("KABOOM!")),
new ProductionExceptionHandlerMock(
-
ProductionExceptionHandler.ProductionExceptionHandlerResponse.CONTINUE,
+ Optional.of(ProductionExceptionHandlerResponse.CONTINUE),
context,
sinkNodeName,
taskId
@@ -1363,12 +1365,12 @@ public class RecordCollectorTest {
public void
shouldThrowStreamsExceptionOnUnknownTopicOrPartitionExceptionWithDefaultExceptionHandler()
{
final KafkaException exception = new TimeoutException("KABOOM!", new
UnknownTopicOrPartitionException());
final RecordCollector collector = new RecordCollectorImpl(
- logContext,
- taskId,
- getExceptionalStreamsProducerOnSend(exception),
- productionExceptionHandler,
- streamsMetrics,
- topology
+ logContext,
+ taskId,
+ getExceptionalStreamsProducerOnSend(exception),
+ productionExceptionHandler,
+ streamsMetrics,
+ topology
);
collector.send(topic, "3", "0", null, null, stringSerializer,
stringSerializer, sinkNodeName, context, streamPartitioner);
@@ -1378,10 +1380,10 @@ public class RecordCollectorTest {
final StreamsException thrown = assertThrows(StreamsException.class,
collector::flush);
assertEquals(exception, thrown.getCause());
assertThat(
- thrown.getMessage(),
- equalTo("Error encountered sending record to topic topic for
task 0_0 due to:" +
- "\norg.apache.kafka.common.errors.TimeoutException:
KABOOM!" +
- "\nException handler choose to FAIL the processing, no
more records would be sent.")
+ thrown.getMessage(),
+ equalTo("Error encountered sending record to topic topic for task
0_0 due to:" +
+ "\norg.apache.kafka.common.errors.TimeoutException:
KABOOM!" +
+ "\nException handler choose to FAIL the processing, no
more records would be sent.")
);
}
@@ -1389,17 +1391,17 @@ public class RecordCollectorTest {
public void
shouldNotThrowTaskCorruptedExceptionOnUnknownTopicOrPartitionExceptionUsingAlwaysContinueExceptionHandler()
{
final KafkaException exception = new TimeoutException("KABOOM!", new
UnknownTopicOrPartitionException());
final RecordCollector collector = new RecordCollectorImpl(
- logContext,
- taskId,
- getExceptionalStreamsProducerOnSend(exception),
- new ProductionExceptionHandlerMock(
-
ProductionExceptionHandler.ProductionExceptionHandlerResponse.CONTINUE,
- context,
- sinkNodeName,
- taskId
- ),
- streamsMetrics,
- topology
+ logContext,
+ taskId,
+ getExceptionalStreamsProducerOnSend(exception),
+ new ProductionExceptionHandlerMock(
+ Optional.of(ProductionExceptionHandlerResponse.CONTINUE),
+ context,
+ sinkNodeName,
+ taskId
+ ),
+ streamsMetrics,
+ topology
);
collector.send(topic, "3", "0", null, null, stringSerializer,
stringSerializer, sinkNodeName, context, streamPartitioner);
@@ -1539,11 +1541,11 @@ public class RecordCollectorTest {
public void shouldDropRecordExceptionUsingAlwaysContinueExceptionHandler()
{
try (final ErrorStringSerializer errorSerializer = new
ErrorStringSerializer()) {
final RecordCollector collector = newRecordCollector(new
ProductionExceptionHandlerMock(
-
ProductionExceptionHandler.ProductionExceptionHandlerResponse.CONTINUE,
+ Optional.of(ProductionExceptionHandlerResponse.CONTINUE),
context,
sinkNodeName,
taskId,
- ProductionExceptionHandler.SerializationExceptionOrigin.KEY
+ SerializationExceptionOrigin.KEY
));
collector.initialize();
@@ -1568,11 +1570,11 @@ public class RecordCollectorTest {
public void
shouldThrowStreamsExceptionWhenValueSerializationFailedAndProductionExceptionHandlerRepliesWithFail()
{
try (final ErrorStringSerializer errorSerializer = new
ErrorStringSerializer()) {
final RecordCollector collector = newRecordCollector(new
ProductionExceptionHandlerMock(
-
ProductionExceptionHandler.ProductionExceptionHandlerResponse.FAIL,
+ Optional.of(ProductionExceptionHandlerResponse.FAIL),
context,
sinkNodeName,
taskId,
- ProductionExceptionHandler.SerializationExceptionOrigin.VALUE
+ SerializationExceptionOrigin.VALUE
));
collector.initialize();
@@ -1589,11 +1591,11 @@ public class RecordCollectorTest {
public void
shouldThrowStreamsExceptionWhenKeySerializationFailedAndProductionExceptionHandlerRepliesWithFail()
{
try (final ErrorStringSerializer errorSerializer = new
ErrorStringSerializer()) {
final RecordCollector collector = newRecordCollector(new
ProductionExceptionHandlerMock(
-
ProductionExceptionHandler.ProductionExceptionHandlerResponse.FAIL,
+ Optional.of(ProductionExceptionHandlerResponse.FAIL),
context,
sinkNodeName,
taskId,
- ProductionExceptionHandler.SerializationExceptionOrigin.KEY
+ SerializationExceptionOrigin.KEY
));
collector.initialize();
@@ -1606,11 +1608,109 @@ public class RecordCollectorTest {
}
}
+ @Test
+ public void
shouldThrowStreamsExceptionWhenSerializationFailedAndProductionExceptionHandlerReturnsNull()
{
+ try (final ErrorStringSerializer errorSerializer = new
ErrorStringSerializer()) {
+ final RecordCollector collector = newRecordCollector(new
ProductionExceptionHandlerMock(
+ Optional.empty(),
+ context,
+ sinkNodeName,
+ taskId,
+ SerializationExceptionOrigin.KEY
+ ));
+ collector.initialize();
+
+ final StreamsException exception = assertThrows(
+ StreamsException.class,
+ () -> collector.send(topic, "key", "val", null, 0, null,
errorSerializer, stringSerializer, sinkNodeName, context)
+ );
+
+ assertEquals("Fatal user code error in production error callback",
exception.getMessage());
+ assertInstanceOf(NullPointerException.class, exception.getCause());
+ assertEquals("Invalid ProductionExceptionHandler response.",
exception.getCause().getMessage());
+ }
+ }
+
+ @Test
+ public void
shouldThrowStreamsExceptionWhenSerializationFailedAndProductionExceptionHandlerThrows()
{
+ try (final ErrorStringSerializer errorSerializer = new
ErrorStringSerializer()) {
+ final RecordCollector collector = newRecordCollector(new
ProductionExceptionHandlerMock(
+ true,
+ context,
+ sinkNodeName,
+ taskId,
+ SerializationExceptionOrigin.KEY
+ ));
+ collector.initialize();
+
+ final StreamsException exception = assertThrows(
+ StreamsException.class,
+ () -> collector.send(topic, "key", "val", null, 0, null,
errorSerializer, stringSerializer, sinkNodeName, context)
+ );
+
+ assertEquals("Fatal user code error in production error callback",
exception.getMessage());
+ assertEquals("CRASH", exception.getCause().getMessage());
+ }
+ }
+
+ @Test
+ public void
shouldThrowStreamsExceptionOnSubsequentFlushIfASendFailsAndProductionExceptionHandlerReturnsNull()
{
+ final KafkaException exception = new KafkaException("KABOOM!");
+ final RecordCollector collector = new RecordCollectorImpl(
+ logContext,
+ taskId,
+ getExceptionalStreamsProducerOnSend(exception),
+ new ProductionExceptionHandlerMock(
+ Optional.empty(),
+ context,
+ sinkNodeName,
+ taskId,
+ SerializationExceptionOrigin.KEY
+ ),
+ streamsMetrics,
+ topology
+ );
+
+ collector.send(topic, "3", "0", null, null, stringSerializer,
stringSerializer, sinkNodeName, context, streamPartitioner);
+
+ final StreamsException thrown = assertThrows(StreamsException.class,
collector::flush);
+ assertEquals("Fatal user code error in production error callback",
thrown.getMessage());
+ assertInstanceOf(NullPointerException.class, thrown.getCause());
+ assertEquals("Invalid ProductionExceptionHandler response.",
thrown.getCause().getMessage());
+ }
+
+ @Test
+ public void
shouldThrowStreamsExceptionOnSubsequentFlushIfASendFailsAndProductionExceptionHandlerThrows()
{
+ final KafkaException exception = new KafkaException("KABOOM!");
+ final RecordCollector collector = new RecordCollectorImpl(
+ logContext,
+ taskId,
+ getExceptionalStreamsProducerOnSend(exception),
+ new ProductionExceptionHandlerMock(
+ true,
+ context,
+ sinkNodeName,
+ taskId,
+ SerializationExceptionOrigin.KEY
+ ),
+ streamsMetrics,
+ topology
+ );
+
+ collector.send(topic, "3", "0", null, null, stringSerializer,
stringSerializer, sinkNodeName, context, streamPartitioner);
+
+ final StreamsException thrown = assertThrows(StreamsException.class,
collector::flush);
+ assertEquals("Fatal user code error in production error callback",
thrown.getMessage());
+ assertEquals("CRASH", thrown.getCause().getMessage());
+ }
+
@SuppressWarnings({"unchecked", "rawtypes"})
@Test
public void shouldNotCallProductionExceptionHandlerOnClassCastException() {
try (final ErrorStringSerializer errorSerializer = new
ErrorStringSerializer()) {
- final RecordCollector collector = newRecordCollector(new
ProductionExceptionHandlerMock(ProductionExceptionHandler.ProductionExceptionHandlerResponse.CONTINUE));
+ final RecordCollector collector = newRecordCollector(
+ new
ProductionExceptionHandlerMock(Optional.of(ProductionExceptionHandlerResponse.CONTINUE))
+ );
collector.initialize();
assertThat(mockProducer.history().isEmpty(), equalTo(true));
@@ -1766,17 +1866,18 @@ public class RecordCollectorTest {
}
public static class ProductionExceptionHandlerMock implements
ProductionExceptionHandler {
- private final ProductionExceptionHandlerResponse response;
+ private final Optional<ProductionExceptionHandlerResponse> response;
+ private boolean shouldThrowException;
private InternalProcessorContext<Void, Void> expectedContext;
private String expectedProcessorNodeId;
private TaskId expectedTaskId;
private SerializationExceptionOrigin
expectedSerializationExceptionOrigin;
- public ProductionExceptionHandlerMock(final
ProductionExceptionHandlerResponse response) {
+ public ProductionExceptionHandlerMock(final
Optional<ProductionExceptionHandlerResponse> response) {
this.response = response;
}
- public ProductionExceptionHandlerMock(final
ProductionExceptionHandlerResponse response,
+ public ProductionExceptionHandlerMock(final
Optional<ProductionExceptionHandlerResponse> response,
final
InternalProcessorContext<Void, Void> context,
final String processorNodeId,
final TaskId taskId) {
@@ -1786,13 +1887,24 @@ public class RecordCollectorTest {
this.expectedTaskId = taskId;
}
- public ProductionExceptionHandlerMock(final
ProductionExceptionHandlerResponse response,
+ public ProductionExceptionHandlerMock(final boolean
shouldThrowException,
+ final
InternalProcessorContext<Void, Void> context,
+ final String processorNodeId,
+ final TaskId taskId,
+ final
SerializationExceptionOrigin origin) {
+ this(Optional.empty(), context, processorNodeId, taskId);
+ this.expectedSerializationExceptionOrigin = origin;
+ this.shouldThrowException = shouldThrowException;
+ }
+
+ public ProductionExceptionHandlerMock(final
Optional<ProductionExceptionHandlerResponse> response,
final
InternalProcessorContext<Void, Void> context,
final String processorNodeId,
final TaskId taskId,
final
SerializationExceptionOrigin origin) {
this(response, context, processorNodeId, taskId);
this.expectedSerializationExceptionOrigin = origin;
+ this.shouldThrowException = false;
}
@Override
@@ -1800,7 +1912,10 @@ public class RecordCollectorTest {
final
ProducerRecord<byte[], byte[]> record,
final Exception
exception) {
assertInputs(context, exception);
- return response;
+ if (shouldThrowException) {
+ throw new RuntimeException("CRASH");
+ }
+ return response.orElse(null);
}
@Override
@@ -1810,7 +1925,10 @@ public class RecordCollectorTest {
final SerializationExceptionOrigin origin) {
assertInputs(context, exception);
assertEquals(expectedSerializationExceptionOrigin, origin);
- return response;
+ if (shouldThrowException) {
+ throw new RuntimeException("CRASH");
+ }
+ return response.orElse(null);
}
@Override
diff --git
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordDeserializerTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordDeserializerTest.java
index 23f364fc6a3..1bca1c9e379 100644
---
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordDeserializerTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordDeserializerTest.java
@@ -25,6 +25,7 @@ import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
+import
org.apache.kafka.streams.errors.DeserializationExceptionHandler.DeserializationHandlerResponse;
import org.apache.kafka.streams.errors.ErrorHandlerContext;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.processor.TaskId;
@@ -62,27 +63,29 @@ public class RecordDeserializerTest {
@Test
public void
shouldReturnConsumerRecordWithDeserializedValueWhenNoExceptions() {
- final RecordDeserializer recordDeserializer = new RecordDeserializer(
- new TheSourceNode(
- sourceNodeName,
- false,
- false,
- "key",
- "value"
- ),
- null,
- new LogContext(),
- new Metrics().sensor("dropped-records")
- );
- final ConsumerRecord<Object, Object> record =
recordDeserializer.deserialize(null, rawRecord);
- assertEquals(rawRecord.topic(), record.topic());
- assertEquals(rawRecord.partition(), record.partition());
- assertEquals(rawRecord.offset(), record.offset());
- assertEquals("key", record.key());
- assertEquals("value", record.value());
- assertEquals(rawRecord.timestamp(), record.timestamp());
- assertEquals(TimestampType.CREATE_TIME, record.timestampType());
- assertEquals(rawRecord.headers(), record.headers());
+ try (final Metrics metrics = new Metrics()) {
+ final RecordDeserializer recordDeserializer = new
RecordDeserializer(
+ new TheSourceNode(
+ sourceNodeName,
+ false,
+ false,
+ "key",
+ "value"
+ ),
+ null,
+ new LogContext(),
+ metrics.sensor("dropped-records")
+ );
+ final ConsumerRecord<Object, Object> record =
recordDeserializer.deserialize(null, rawRecord);
+ assertEquals(rawRecord.topic(), record.topic());
+ assertEquals(rawRecord.partition(), record.partition());
+ assertEquals(rawRecord.offset(), record.offset());
+ assertEquals("key", record.key());
+ assertEquals("value", record.value());
+ assertEquals(rawRecord.timestamp(), record.timestamp());
+ assertEquals(TimestampType.CREATE_TIME, record.timestampType());
+ assertEquals(rawRecord.headers(), record.headers());
+ }
}
@ParameterizedTest
@@ -93,30 +96,35 @@ public class RecordDeserializerTest {
})
public void
shouldThrowStreamsExceptionWhenDeserializationFailsAndExceptionHandlerRepliesWithFail(final
boolean keyThrowsException,
final boolean valueThrowsException) {
- final RecordDeserializer recordDeserializer = new RecordDeserializer(
- new TheSourceNode(
- sourceNodeName,
- keyThrowsException,
- valueThrowsException,
- "key",
- "value"
- ),
- new DeserializationExceptionHandlerMock(
-
DeserializationExceptionHandler.DeserializationHandlerResponse.FAIL,
- rawRecord,
- sourceNodeName,
- taskId
- ),
- new LogContext(),
- new Metrics().sensor("dropped-records")
- );
-
- final StreamsException e = assertThrows(StreamsException.class, () ->
recordDeserializer.deserialize(context, rawRecord));
- assertEquals(e.getMessage(), "Deserialization exception handler is set
"
- + "to fail upon a deserialization error. "
- + "If you would rather have the streaming pipeline "
- + "continue after a deserialization error, please set the "
- + "default.deserialization.exception.handler appropriately.");
+ try (final Metrics metrics = new Metrics()) {
+ final RecordDeserializer recordDeserializer = new
RecordDeserializer(
+ new TheSourceNode(
+ sourceNodeName,
+ keyThrowsException,
+ valueThrowsException,
+ "key",
+ "value"
+ ),
+ new DeserializationExceptionHandlerMock(
+ Optional.of(DeserializationHandlerResponse.FAIL),
+ rawRecord,
+ sourceNodeName,
+ taskId
+ ),
+ new LogContext(),
+ metrics.sensor("dropped-records")
+ );
+
+ final StreamsException e = assertThrows(StreamsException.class, ()
-> recordDeserializer.deserialize(context, rawRecord));
+ assertEquals(
+ e.getMessage(),
+ "Deserialization exception handler is set "
+ + "to fail upon a deserialization error. "
+ + "If you would rather have the streaming pipeline
"
+ + "continue after a deserialization error, please
set the "
+ + "default.deserialization.exception.handler
appropriately."
+ );
+ }
}
@ParameterizedTest
@@ -127,26 +135,89 @@ public class RecordDeserializerTest {
})
public void
shouldNotThrowStreamsExceptionWhenDeserializationFailsAndExceptionHandlerRepliesWithContinue(final
boolean keyThrowsException,
final boolean valueThrowsException) {
- final RecordDeserializer recordDeserializer = new RecordDeserializer(
- new TheSourceNode(
- sourceNodeName,
- keyThrowsException,
- valueThrowsException,
- "key",
- "value"
- ),
- new DeserializationExceptionHandlerMock(
-
DeserializationExceptionHandler.DeserializationHandlerResponse.CONTINUE,
- rawRecord,
- sourceNodeName,
- taskId
- ),
- new LogContext(),
- new Metrics().sensor("dropped-records")
- );
-
- final ConsumerRecord<Object, Object> record =
recordDeserializer.deserialize(context, rawRecord);
- assertNull(record);
+ try (final Metrics metrics = new Metrics()) {
+ final RecordDeserializer recordDeserializer = new
RecordDeserializer(
+ new TheSourceNode(
+ sourceNodeName,
+ keyThrowsException,
+ valueThrowsException,
+ "key",
+ "value"
+ ),
+ new DeserializationExceptionHandlerMock(
+
Optional.of(DeserializationHandlerResponse.CONTINUE),
+ rawRecord,
+ sourceNodeName,
+ taskId
+ ),
+ new LogContext(),
+ metrics.sensor("dropped-records")
+ );
+
+ final ConsumerRecord<Object, Object> record =
recordDeserializer.deserialize(context, rawRecord);
+ assertNull(record);
+ }
+ }
+
+ @Test
+ public void
shouldFailWhenDeserializationFailsAndExceptionHandlerReturnsNull() {
+ try (final Metrics metrics = new Metrics()) {
+ final RecordDeserializer recordDeserializer = new
RecordDeserializer(
+ new TheSourceNode(
+ sourceNodeName,
+ true,
+ false,
+ "key",
+ "value"
+ ),
+ new DeserializationExceptionHandlerMock(
+ Optional.empty(),
+ rawRecord,
+ sourceNodeName,
+ taskId
+ ),
+ new LogContext(),
+ metrics.sensor("dropped-records")
+ );
+
+ final StreamsException exception = assertThrows(
+ StreamsException.class,
+ () -> recordDeserializer.deserialize(context, rawRecord)
+ );
+ assertEquals("Fatal user code error in deserialization error
callback", exception.getMessage());
+ assertInstanceOf(NullPointerException.class, exception.getCause());
+ assertEquals("Invalid DeserializationExceptionHandler response.",
exception.getCause().getMessage());
+ }
+ }
+
+ @Test
+ public void shouldFailWhenDeserializationFailsAndExceptionHandlerThrows() {
+ try (final Metrics metrics = new Metrics()) {
+ final RecordDeserializer recordDeserializer = new
RecordDeserializer(
+ new TheSourceNode(
+ sourceNodeName,
+ true,
+ false,
+ "key",
+ "value"
+ ),
+ new DeserializationExceptionHandlerMock(
+ null, // indicate to throw an exception
+ rawRecord,
+ sourceNodeName,
+ taskId
+ ),
+ new LogContext(),
+ metrics.sensor("dropped-records")
+ );
+
+ final StreamsException exception = assertThrows(
+ StreamsException.class,
+ () -> recordDeserializer.deserialize(context, rawRecord)
+ );
+ assertEquals("Fatal user code error in deserialization error
callback", exception.getMessage());
+ assertEquals("CRASH", exception.getCause().getMessage());
+ }
}
static class TheSourceNode extends SourceNode<Object, Object> {
@@ -185,12 +256,12 @@ public class RecordDeserializerTest {
}
public static class DeserializationExceptionHandlerMock implements
DeserializationExceptionHandler {
- private final DeserializationHandlerResponse response;
+ private final Optional<DeserializationHandlerResponse> response;
private final ConsumerRecord<byte[], byte[]> expectedRecord;
private final String expectedProcessorNodeId;
private final TaskId expectedTaskId;
- public DeserializationExceptionHandlerMock(final
DeserializationHandlerResponse response,
+ public DeserializationExceptionHandlerMock(final
Optional<DeserializationHandlerResponse> response,
final
ConsumerRecord<byte[], byte[]> record,
final String
processorNodeId,
final TaskId taskId) {
@@ -212,7 +283,10 @@ public class RecordDeserializerTest {
assertEquals(expectedRecord, record);
assertInstanceOf(RuntimeException.class, exception);
assertEquals("KABOOM!", exception.getMessage());
- return response;
+ if (response == null) {
+ throw new RuntimeException("CRASH");
+ }
+ return response.orElse(null);
}
@Override
diff --git
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
index a8771c21539..3fa33ef8954 100644
---
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
@@ -53,6 +53,7 @@ import
org.apache.kafka.streams.errors.LogAndContinueProcessingExceptionHandler;
import org.apache.kafka.streams.errors.LogAndFailExceptionHandler;
import org.apache.kafka.streams.errors.LogAndFailProcessingExceptionHandler;
import org.apache.kafka.streams.errors.ProcessingExceptionHandler;
+import
org.apache.kafka.streams.errors.ProcessingExceptionHandler.ProcessingHandlerResponse;
import org.apache.kafka.streams.errors.ProcessorStateException;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.errors.TaskCorruptedException;
@@ -2661,17 +2662,20 @@ public class StreamTaskTest {
}
@Test
- public void
shouldPunctuateNotHandleFailProcessingExceptionAndThrowStreamsException() {
+ public void
punctuateShouldNotHandleFailProcessingExceptionAndThrowStreamsException() {
when(stateManager.taskId()).thenReturn(taskId);
when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
- task = createStatelessTask(createConfig(AT_LEAST_ONCE, "100",
- LogAndFailExceptionHandler.class.getName(),
LogAndContinueProcessingExceptionHandler.class.getName()));
+ task = createStatelessTask(createConfig(
+ AT_LEAST_ONCE,
+ "100",
+ LogAndFailExceptionHandler.class.getName(),
+ LogAndContinueProcessingExceptionHandler.class.getName()
+ ));
- final StreamsException streamsException =
assertThrows(StreamsException.class, () ->
- task.punctuate(processorStreamTime, 1,
PunctuationType.STREAM_TIME, timestamp -> {
- throw new FailedProcessingException(
- new RuntimeException("KABOOM!")
- );
+ final StreamsException streamsException = assertThrows(
+ StreamsException.class,
+ () -> task.punctuate(processorStreamTime, 1,
PunctuationType.STREAM_TIME, timestamp -> {
+ throw new FailedProcessingException(new
RuntimeException("KABOOM!"));
})
);
@@ -2680,11 +2684,15 @@ public class StreamTaskTest {
}
@Test
- public void shouldPunctuateNotHandleTaskCorruptedExceptionAndThrowItAsIs()
{
+ public void punctuateShouldNotHandleTaskCorruptedExceptionAndThrowItAsIs()
{
when(stateManager.taskId()).thenReturn(taskId);
when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
- task = createStatelessTask(createConfig(AT_LEAST_ONCE, "100",
- LogAndFailExceptionHandler.class.getName(),
LogAndContinueProcessingExceptionHandler.class.getName()));
+ task = createStatelessTask(createConfig(
+ AT_LEAST_ONCE,
+ "100",
+ LogAndFailExceptionHandler.class.getName(),
+ LogAndContinueProcessingExceptionHandler.class.getName()
+ ));
final Set<TaskId> tasksIds = new HashSet<>();
tasksIds.add(new TaskId(0, 0));
@@ -2695,8 +2703,9 @@ public class StreamTaskTest {
}
});
- final TaskCorruptedException taskCorruptedException =
assertThrows(TaskCorruptedException.class, () ->
- task.punctuate(processorStreamTime, 1,
PunctuationType.STREAM_TIME, timestamp -> {
+ final TaskCorruptedException taskCorruptedException = assertThrows(
+ TaskCorruptedException.class,
+ () -> task.punctuate(processorStreamTime, 1,
PunctuationType.STREAM_TIME, timestamp -> {
throw expectedException;
})
);
@@ -2705,16 +2714,21 @@ public class StreamTaskTest {
}
@Test
- public void shouldPunctuateNotHandleTaskMigratedExceptionAndThrowItAsIs() {
+ public void punctuateShouldNotHandleTaskMigratedExceptionAndThrowItAsIs() {
when(stateManager.taskId()).thenReturn(taskId);
when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
- task = createStatelessTask(createConfig(AT_LEAST_ONCE, "100",
- LogAndFailExceptionHandler.class.getName(),
LogAndContinueProcessingExceptionHandler.class.getName()));
+ task = createStatelessTask(createConfig(
+ AT_LEAST_ONCE,
+ "100",
+ LogAndFailExceptionHandler.class.getName(),
+ LogAndContinueProcessingExceptionHandler.class.getName()
+ ));
final TaskMigratedException expectedException = new
TaskMigratedException("TaskMigratedException", new RuntimeException("Task
migrated cause"));
- final TaskMigratedException taskCorruptedException =
assertThrows(TaskMigratedException.class, () ->
- task.punctuate(processorStreamTime, 1,
PunctuationType.STREAM_TIME, timestamp -> {
+ final TaskMigratedException taskCorruptedException = assertThrows(
+ TaskMigratedException.class,
+ () -> task.punctuate(processorStreamTime, 1,
PunctuationType.STREAM_TIME, timestamp -> {
throw expectedException;
})
);
@@ -2723,56 +2737,106 @@ public class StreamTaskTest {
}
@Test
- public void
shouldPunctuateNotThrowStreamsExceptionWhenProcessingExceptionHandlerRepliesWithContinue()
{
+ public void
punctuateShouldNotThrowStreamsExceptionWhenProcessingExceptionHandlerRepliesWithContinue()
{
when(stateManager.taskId()).thenReturn(taskId);
when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
- task = createStatelessTask(createConfig(AT_LEAST_ONCE, "100",
- LogAndFailExceptionHandler.class.getName(),
LogAndContinueProcessingExceptionHandler.class.getName()));
+ task = createStatelessTask(createConfig(
+ AT_LEAST_ONCE,
+ "100",
+ LogAndFailExceptionHandler.class.getName(),
+ LogAndContinueProcessingExceptionHandler.class.getName()
+ ));
- assertDoesNotThrow(() ->
- task.punctuate(processorStreamTime, 1,
PunctuationType.STREAM_TIME, timestamp -> {
+ task.punctuate(processorStreamTime, 1, PunctuationType.STREAM_TIME,
timestamp -> {
+ throw new KafkaException("KABOOM!");
+ });
+ }
+
+ @Test
+ public void
punctuateShouldThrowStreamsExceptionWhenProcessingExceptionHandlerRepliesWithFail()
{
+ when(stateManager.taskId()).thenReturn(taskId);
+ when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
+ task = createStatelessTask(createConfig(
+ AT_LEAST_ONCE,
+ "100",
+ LogAndFailExceptionHandler.class.getName(),
+ LogAndFailProcessingExceptionHandler.class.getName()
+ ));
+
+ final StreamsException streamsException = assertThrows(
+ StreamsException.class,
+ () -> task.punctuate(processorStreamTime, 1,
PunctuationType.STREAM_TIME, timestamp -> {
throw new KafkaException("KABOOM!");
})
);
+
+ assertInstanceOf(KafkaException.class, streamsException.getCause());
+ assertEquals("KABOOM!", streamsException.getCause().getMessage());
}
@Test
- public void
shouldPunctuateThrowStreamsExceptionWhenProcessingExceptionHandlerRepliesWithFail()
{
+ public void
punctuateShouldThrowStreamsExceptionWhenProcessingExceptionHandlerReturnsNull()
{
when(stateManager.taskId()).thenReturn(taskId);
when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
- task = createStatelessTask(createConfig(AT_LEAST_ONCE, "100",
- LogAndFailExceptionHandler.class.getName(),
LogAndFailProcessingExceptionHandler.class.getName()));
+ task = createStatelessTask(createConfig(
+ AT_LEAST_ONCE,
+ "100",
+ LogAndFailExceptionHandler.class.getName(),
+ NullProcessingExceptionHandler.class.getName()
+ ));
- final StreamsException streamsException =
assertThrows(StreamsException.class,
+ final StreamsException streamsException = assertThrows(
+ StreamsException.class,
() -> task.punctuate(processorStreamTime, 1,
PunctuationType.STREAM_TIME, timestamp -> {
throw new KafkaException("KABOOM!");
- }));
+ })
+ );
- assertInstanceOf(KafkaException.class, streamsException.getCause());
- assertEquals("KABOOM!", streamsException.getCause().getMessage());
+ assertEquals("Fatal user code error in processing error callback",
streamsException.getMessage());
+ assertInstanceOf(NullPointerException.class,
streamsException.getCause());
+ assertEquals("Invalid ProcessingExceptionHandler response.",
streamsException.getCause().getMessage());
}
@Test
- public void
shouldPunctuateThrowFailedProcessingExceptionWhenProcessingExceptionHandlerThrowsAnException()
{
+ public void
punctuateShouldThrowFailedProcessingExceptionWhenProcessingExceptionHandlerThrowsAnException()
{
when(stateManager.taskId()).thenReturn(taskId);
when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
- task = createStatelessTask(createConfig(AT_LEAST_ONCE, "100",
- LogAndFailExceptionHandler.class.getName(),
ProcessingExceptionHandlerMock.class.getName()));
+ task = createStatelessTask(createConfig(
+ AT_LEAST_ONCE,
+ "100",
+ LogAndFailExceptionHandler.class.getName(),
+ CrashingProcessingExceptionHandler.class.getName()
+ ));
- final FailedProcessingException streamsException =
assertThrows(FailedProcessingException.class,
+ final FailedProcessingException streamsException = assertThrows(
+ FailedProcessingException.class,
() -> task.punctuate(processorStreamTime, 1,
PunctuationType.STREAM_TIME, timestamp -> {
throw new KafkaException("KABOOM!");
- }));
+ })
+ );
- assertInstanceOf(RuntimeException.class, streamsException.getCause());
+ assertEquals("Fatal user code error in processing error callback",
streamsException.getMessage());
assertEquals("KABOOM from ProcessingExceptionHandlerMock!",
streamsException.getCause().getMessage());
}
- public static class ProcessingExceptionHandlerMock implements
ProcessingExceptionHandler {
+ public static class CrashingProcessingExceptionHandler implements
ProcessingExceptionHandler {
@Override
- public ProcessingExceptionHandler.ProcessingHandlerResponse
handle(final ErrorHandlerContext context, final Record<?, ?> record, final
Exception exception) {
+ public ProcessingHandlerResponse handle(final ErrorHandlerContext
context, final Record<?, ?> record, final Exception exception) {
throw new RuntimeException("KABOOM from
ProcessingExceptionHandlerMock!");
}
+
+ @Override
+ public void configure(final Map<String, ?> configs) {
+ // No-op
+ }
+ }
+
+ public static class NullProcessingExceptionHandler implements
ProcessingExceptionHandler {
+ @Override
+ public ProcessingHandlerResponse handle(final ErrorHandlerContext
context, final Record<?, ?> record, final Exception exception) {
+ return null;
+ }
+
@Override
public void configure(final Map<String, ?> configs) {
// No-op
