You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mj...@apache.org on 2024/07/30 04:22:41 UTC
(kafka) branch 3.9 updated (f26f0b66261 -> b8532070f79)
This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a change to branch 3.9
in repository https://gitbox.apache.org/repos/asf/kafka.git
from f26f0b66261 tests/kafkatest/version.py: Add 3.9.0 as DEV_VERSION
new a4ea9aec73a KAFKA-16448: Add ErrorHandlerContext in production exception handler (#16433)
new 10d9f7872d8 KAFKA-16448: Add ErrorHandlerContext in deserialization exception handler (#16432)
new b8532070f79 HOTFIX: fix compilation error
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../errors/DefaultProductionExceptionHandler.java | 8 +
.../errors/DeserializationExceptionHandler.java | 26 ++-
.../kafka/streams/errors/ErrorHandlerContext.java | 28 ---
.../errors/LogAndContinueExceptionHandler.java | 14 ++
.../streams/errors/LogAndFailExceptionHandler.java | 14 ++
.../streams/errors/ProductionExceptionHandler.java | 49 +++++-
.../internals/DefaultErrorHandlerContext.java | 28 ++-
.../processor/internals/CorruptedRecord.java | 2 +-
.../internals/GlobalStateManagerImpl.java | 3 +-
.../processor/internals/GlobalStateUpdateTask.java | 3 +-
.../processor/internals/ProcessorAdapter.java | 3 +-
.../processor/internals/ProcessorContextImpl.java | 3 +-
.../streams/processor/internals/ProcessorNode.java | 3 +-
.../internals/ProcessorRecordContext.java | 18 +-
.../processor/internals/RecordCollectorImpl.java | 192 +++++++++++++++------
.../processor/internals/RecordDeserializer.java | 24 ++-
.../streams/processor/internals/RecordQueue.java | 2 +-
.../streams/processor/internals/SinkNode.java | 3 +-
.../streams/processor/internals/StampedRecord.java | 18 +-
.../streams/processor/internals/StreamTask.java | 6 +-
.../AlwaysContinueProductionExceptionHandler.java | 44 -----
.../ProcessingExceptionHandlerIntegrationTest.java | 2 -
.../processor/internals/ProcessorNodeTest.java | 7 +-
.../processor/internals/RecordCollectorTest.java | 187 ++++++++++++++++----
.../internals/RecordDeserializerTest.java | 131 +++++++++++++-
.../kafka/test/InternalMockProcessorContext.java | 7 +
26 files changed, 576 insertions(+), 249 deletions(-)
delete mode 100644 streams/src/test/java/org/apache/kafka/streams/errors/AlwaysContinueProductionExceptionHandler.java
(kafka) 02/03: KAFKA-16448: Add ErrorHandlerContext in deserialization exception handler (#16432)
Posted by mj...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch 3.9
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit 10d9f7872d8fad81cd95f327ba6140994cf35576
Author: Sebastien Viale <se...@michelin.com>
AuthorDate: Tue Jul 30 05:33:33 2024 +0200
KAFKA-16448: Add ErrorHandlerContext in deserialization exception handler (#16432)
This PR is part of KIP1033 which aims to bring a ProcessingExceptionHandler to Kafka Streams in order to deal with exceptions that occur during processing.
This PR expose the new ErrorHandlerContext as a parameter to the Deserialization exception handlers and deprecate the previous handle signature.
Co-authored-by: Dabz <d....@gmail.com>
Co-authored-by: loicgreffier <lo...@michelin.com>
Reviewers: Bruno Cadonna <br...@confluent.io>, Matthias J. Sax <ma...@confluent.io>
---
.../errors/DeserializationExceptionHandler.java | 26 +++-
.../errors/LogAndContinueExceptionHandler.java | 14 +++
.../streams/errors/LogAndFailExceptionHandler.java | 14 +++
.../internals/DefaultErrorHandlerContext.java | 12 +-
.../streams/processor/internals/ProcessorNode.java | 1 +
.../processor/internals/RecordDeserializer.java | 24 +++-
.../internals/RecordDeserializerTest.java | 131 +++++++++++++++++++--
7 files changed, 205 insertions(+), 17 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 95ccfeced8e..0d64611de67 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
@@ -16,9 +16,9 @@
*/
package org.apache.kafka.streams.errors;
-
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.Configurable;
+import org.apache.kafka.streams.errors.internals.DefaultErrorHandlerContext;
import org.apache.kafka.streams.processor.ProcessorContext;
/**
@@ -37,11 +37,27 @@ 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
+ default DeserializationHandlerResponse handle(final ProcessorContext context,
+ final ConsumerRecord<byte[], byte[]> record,
+ final Exception exception) {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * Inspect a record and the exception received.
+ *
+ * @param context error handler context
+ * @param record record that failed deserialization
+ * @param exception the actual exception
*/
- @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
- DeserializationHandlerResponse handle(final ProcessorContext context,
- final ConsumerRecord<byte[], byte[]> record,
- final Exception exception);
+ default DeserializationHandlerResponse handle(final ErrorHandlerContext context,
+ final ConsumerRecord<byte[], byte[]> record,
+ final Exception exception) {
+ return handle(((DefaultErrorHandlerContext) context).processorContext().orElse(null), record, exception);
+ }
/**
* Enumeration that describes the response from the exception handler.
diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/LogAndContinueExceptionHandler.java b/streams/src/main/java/org/apache/kafka/streams/errors/LogAndContinueExceptionHandler.java
index a468be2e67a..a93b7c99517 100644
--- a/streams/src/main/java/org/apache/kafka/streams/errors/LogAndContinueExceptionHandler.java
+++ b/streams/src/main/java/org/apache/kafka/streams/errors/LogAndContinueExceptionHandler.java
@@ -32,6 +32,7 @@ import java.util.Map;
public class LogAndContinueExceptionHandler implements DeserializationExceptionHandler {
private static final Logger log = LoggerFactory.getLogger(LogAndContinueExceptionHandler.class);
+ @Deprecated
@Override
public DeserializationHandlerResponse handle(final ProcessorContext context,
final ConsumerRecord<byte[], byte[]> record,
@@ -45,6 +46,19 @@ public class LogAndContinueExceptionHandler implements DeserializationExceptionH
return DeserializationHandlerResponse.CONTINUE;
}
+ @Override
+ public DeserializationHandlerResponse handle(final ErrorHandlerContext context,
+ final ConsumerRecord<byte[], byte[]> record,
+ final Exception exception) {
+
+ log.warn("Exception caught during Deserialization, " +
+ "taskId: {}, topic: {}, partition: {}, offset: {}",
+ context.taskId(), record.topic(), record.partition(), record.offset(),
+ exception);
+
+ return DeserializationHandlerResponse.CONTINUE;
+ }
+
@Override
public void configure(final Map<String, ?> configs) {
// ignore
diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/LogAndFailExceptionHandler.java b/streams/src/main/java/org/apache/kafka/streams/errors/LogAndFailExceptionHandler.java
index 3f0fd486520..5fdda623bdd 100644
--- a/streams/src/main/java/org/apache/kafka/streams/errors/LogAndFailExceptionHandler.java
+++ b/streams/src/main/java/org/apache/kafka/streams/errors/LogAndFailExceptionHandler.java
@@ -33,6 +33,7 @@ public class LogAndFailExceptionHandler implements DeserializationExceptionHandl
private static final Logger log = LoggerFactory.getLogger(LogAndFailExceptionHandler.class);
@Override
+ @Deprecated
public DeserializationHandlerResponse handle(final ProcessorContext context,
final ConsumerRecord<byte[], byte[]> record,
final Exception exception) {
@@ -45,6 +46,19 @@ public class LogAndFailExceptionHandler implements DeserializationExceptionHandl
return DeserializationHandlerResponse.FAIL;
}
+ @Override
+ public DeserializationHandlerResponse handle(final ErrorHandlerContext context,
+ final ConsumerRecord<byte[], byte[]> record,
+ final Exception exception) {
+
+ log.error("Exception caught during Deserialization, " +
+ "taskId: {}, topic: {}, partition: {}, offset: {}",
+ context.taskId(), record.topic(), record.partition(), record.offset(),
+ exception);
+
+ return DeserializationHandlerResponse.FAIL;
+ }
+
@Override
public void configure(final Map<String, ?> configs) {
// ignore
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 aa066fb6da9..c907ff3eb89 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
@@ -18,8 +18,11 @@ package org.apache.kafka.streams.errors.internals;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.streams.errors.ErrorHandlerContext;
+import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.TaskId;
+import java.util.Optional;
+
/**
* Default implementation of {@link ErrorHandlerContext} that provides access to the metadata of the record that caused the error.
*/
@@ -30,8 +33,10 @@ public class DefaultErrorHandlerContext implements ErrorHandlerContext {
private final Headers headers;
private final String processorNodeId;
private final TaskId taskId;
+ private ProcessorContext processorContext;
- public DefaultErrorHandlerContext(final String topic,
+ public DefaultErrorHandlerContext(final ProcessorContext processorContext,
+ final String topic,
final int partition,
final long offset,
final Headers headers,
@@ -43,6 +48,7 @@ public class DefaultErrorHandlerContext implements ErrorHandlerContext {
this.headers = headers;
this.processorNodeId = processorNodeId;
this.taskId = taskId;
+ this.processorContext = processorContext;
}
@Override
@@ -74,4 +80,8 @@ public class DefaultErrorHandlerContext implements ErrorHandlerContext {
public TaskId taskId() {
return taskId;
}
+
+ public Optional<ProcessorContext> processorContext() {
+ return Optional.ofNullable(processorContext);
+ }
}
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 c30d42ba746..65eec47cb1d 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
@@ -205,6 +205,7 @@ public class ProcessorNode<KIn, VIn, KOut, VOut> {
throw e;
} catch (final Exception e) {
final ErrorHandlerContext errorHandlerContext = new DefaultErrorHandlerContext(
+ null,
internalProcessorContext.topic(),
internalProcessorContext.partition(),
internalProcessorContext.offset(),
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 2680c179b54..8ee2dc014eb 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
@@ -22,6 +22,7 @@ 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.StreamsException;
+import org.apache.kafka.streams.errors.internals.DefaultErrorHandlerContext;
import org.apache.kafka.streams.processor.api.ProcessorContext;
import org.slf4j.Logger;
@@ -69,7 +70,7 @@ public class RecordDeserializer {
Optional.empty()
);
} catch (final Exception deserializationException) {
- handleDeserializationFailure(deserializationExceptionHandler, processorContext, deserializationException, rawRecord, log, droppedRecordsSensor);
+ 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'
}
}
@@ -80,12 +81,27 @@ public class RecordDeserializer {
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 ConsumerRecord<byte[], byte[]> rawRecord,
+ final Logger log,
+ final Sensor droppedRecordsSensor,
+ final String sourceNodeName) {
final DeserializationExceptionHandler.DeserializationHandlerResponse response;
try {
- response = deserializationExceptionHandler.handle(
+ final DefaultErrorHandlerContext errorHandlerContext = new DefaultErrorHandlerContext(
(InternalProcessorContext<?, ?>) processorContext,
- rawRecord,
- deserializationException);
+ rawRecord.topic(),
+ rawRecord.partition(),
+ rawRecord.offset(),
+ rawRecord.headers(),
+ sourceNodeName,
+ processorContext.taskId());
+ response = deserializationExceptionHandler.handle(errorHandlerContext, rawRecord, deserializationException);
} catch (final Exception fatalUserException) {
log.error(
"Deserialization error callback failed after deserialization error for record {}",
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 ac973c1789d..23f364fc6a3 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
@@ -24,16 +24,28 @@ import org.apache.kafka.common.header.internals.RecordHeaders;
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.ErrorHandlerContext;
+import org.apache.kafka.streams.errors.StreamsException;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.test.InternalMockProcessorContext;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import java.util.Map;
import java.util.Optional;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
public class RecordDeserializerTest {
-
- private final RecordHeaders headers = new RecordHeaders(new Header[] {new RecordHeader("key", "value".getBytes())});
+ private final String sourceNodeName = "source-node";
+ private final TaskId taskId = new TaskId(0, 0);
+ private final RecordHeaders headers = new RecordHeaders(new Header[]{new RecordHeader("key", "value".getBytes())});
private final ConsumerRecord<byte[], byte[]> rawRecord = new ConsumerRecord<>("topic",
1,
1,
@@ -46,13 +58,17 @@ public class RecordDeserializerTest {
headers,
Optional.empty());
+ private final InternalProcessorContext<Void, Void> context = new InternalMockProcessorContext<>();
+
@Test
public void shouldReturnConsumerRecordWithDeserializedValueWhenNoExceptions() {
final RecordDeserializer recordDeserializer = new RecordDeserializer(
new TheSourceNode(
+ sourceNodeName,
false,
false,
- "key", "value"
+ "key",
+ "value"
),
null,
new LogContext(),
@@ -69,17 +85,82 @@ public class RecordDeserializerTest {
assertEquals(rawRecord.headers(), record.headers());
}
+ @ParameterizedTest
+ @CsvSource({
+ "true, true",
+ "true, false",
+ "false, true",
+ })
+ 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.");
+ }
+
+ @ParameterizedTest
+ @CsvSource({
+ "true, true",
+ "true, false",
+ "false, true"
+ })
+ 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);
+ }
+
static class TheSourceNode extends SourceNode<Object, Object> {
private final boolean keyThrowsException;
private final boolean valueThrowsException;
private final Object key;
private final Object value;
- TheSourceNode(final boolean keyThrowsException,
+ TheSourceNode(final String name,
+ final boolean keyThrowsException,
final boolean valueThrowsException,
final Object key,
final Object value) {
- super("", null, null);
+ super(name, null, null);
this.keyThrowsException = keyThrowsException;
this.valueThrowsException = valueThrowsException;
this.key = key;
@@ -89,7 +170,7 @@ public class RecordDeserializerTest {
@Override
public Object deserializeKey(final String topic, final Headers headers, final byte[] data) {
if (keyThrowsException) {
- throw new RuntimeException();
+ throw new RuntimeException("KABOOM!");
}
return key;
}
@@ -97,10 +178,46 @@ public class RecordDeserializerTest {
@Override
public Object deserializeValue(final String topic, final Headers headers, final byte[] data) {
if (valueThrowsException) {
- throw new RuntimeException();
+ throw new RuntimeException("KABOOM!");
}
return value;
}
}
+ public static class DeserializationExceptionHandlerMock implements DeserializationExceptionHandler {
+ private final DeserializationHandlerResponse response;
+ private final ConsumerRecord<byte[], byte[]> expectedRecord;
+ private final String expectedProcessorNodeId;
+ private final TaskId expectedTaskId;
+
+ public DeserializationExceptionHandlerMock(final DeserializationHandlerResponse response,
+ final ConsumerRecord<byte[], byte[]> record,
+ final String processorNodeId,
+ final TaskId taskId) {
+ this.response = response;
+ this.expectedRecord = record;
+ this.expectedProcessorNodeId = processorNodeId;
+ this.expectedTaskId = taskId;
+ }
+
+ @Override
+ public DeserializationHandlerResponse handle(final ErrorHandlerContext context,
+ final ConsumerRecord<byte[], byte[]> record,
+ final Exception exception) {
+ assertEquals(expectedRecord.topic(), context.topic());
+ assertEquals(expectedRecord.partition(), context.partition());
+ assertEquals(expectedRecord.offset(), context.offset());
+ assertEquals(expectedProcessorNodeId, context.processorNodeId());
+ assertEquals(expectedTaskId, context.taskId());
+ assertEquals(expectedRecord, record);
+ assertInstanceOf(RuntimeException.class, exception);
+ assertEquals("KABOOM!", exception.getMessage());
+ return response;
+ }
+
+ @Override
+ public void configure(final Map<String, ?> configs) {
+ // do nothing
+ }
+ }
}
(kafka) 01/03: KAFKA-16448: Add ErrorHandlerContext in production exception handler (#16433)
Posted by mj...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch 3.9
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit a4ea9aec73a3da470d079bad82d92810cac49d55
Author: Sebastien Viale <se...@michelin.com>
AuthorDate: Tue Jul 30 05:17:15 2024 +0200
KAFKA-16448: Add ErrorHandlerContext in production exception handler (#16433)
This PR is part of KIP-1033 which aims to bring a ProcessingExceptionHandler to Kafka Streams in order to deal with exceptions that occur during processing.
This PR expose the new ErrorHandlerContext as a parameter to the Production exception handler and deprecate the previous handle signature.
Co-authored-by: Dabz <d....@gmail.com>
Co-authored-by: loicgreffier <lo...@michelin.com>
Reviewers: Bruno Cadonna <br...@confluent.io>, Matthias J. Sax <ma...@confluent.io>
---
.../errors/DefaultProductionExceptionHandler.java | 8 +
.../kafka/streams/errors/ErrorHandlerContext.java | 28 ---
.../streams/errors/ProductionExceptionHandler.java | 49 +++++-
.../internals/DefaultErrorHandlerContext.java | 16 --
.../processor/internals/CorruptedRecord.java | 2 +-
.../internals/GlobalStateManagerImpl.java | 3 +-
.../processor/internals/GlobalStateUpdateTask.java | 3 +-
.../processor/internals/ProcessorAdapter.java | 3 +-
.../processor/internals/ProcessorContextImpl.java | 3 +-
.../streams/processor/internals/ProcessorNode.java | 2 -
.../internals/ProcessorRecordContext.java | 18 +-
.../processor/internals/RecordCollectorImpl.java | 188 +++++++++++++++------
.../streams/processor/internals/RecordQueue.java | 2 +-
.../streams/processor/internals/SinkNode.java | 3 +-
.../streams/processor/internals/StampedRecord.java | 18 +-
.../streams/processor/internals/StreamTask.java | 6 +-
.../AlwaysContinueProductionExceptionHandler.java | 44 -----
.../ProcessingExceptionHandlerIntegrationTest.java | 2 -
.../processor/internals/ProcessorNodeTest.java | 7 +-
.../processor/internals/RecordCollectorTest.java | 187 ++++++++++++++++----
.../kafka/test/InternalMockProcessorContext.java | 7 +
21 files changed, 367 insertions(+), 232 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/DefaultProductionExceptionHandler.java b/streams/src/main/java/org/apache/kafka/streams/errors/DefaultProductionExceptionHandler.java
index 33a95f4b445..0896114cf28 100644
--- a/streams/src/main/java/org/apache/kafka/streams/errors/DefaultProductionExceptionHandler.java
+++ b/streams/src/main/java/org/apache/kafka/streams/errors/DefaultProductionExceptionHandler.java
@@ -25,12 +25,20 @@ import java.util.Map;
* happens while attempting to produce result records.
*/
public class DefaultProductionExceptionHandler implements ProductionExceptionHandler {
+ @Deprecated
@Override
public ProductionExceptionHandlerResponse handle(final ProducerRecord<byte[], byte[]> record,
final Exception exception) {
return ProductionExceptionHandlerResponse.FAIL;
}
+ @Override
+ public ProductionExceptionHandlerResponse handle(final ErrorHandlerContext context,
+ final ProducerRecord<byte[], byte[]> record,
+ final Exception exception) {
+ return ProductionExceptionHandlerResponse.FAIL;
+ }
+
@Override
public void configure(final Map<String, ?> configs) {
// ignore
diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/ErrorHandlerContext.java b/streams/src/main/java/org/apache/kafka/streams/errors/ErrorHandlerContext.java
index 0c505475490..6c5e4f19596 100644
--- a/streams/src/main/java/org/apache/kafka/streams/errors/ErrorHandlerContext.java
+++ b/streams/src/main/java/org/apache/kafka/streams/errors/ErrorHandlerContext.java
@@ -89,34 +89,6 @@ public interface ErrorHandlerContext {
*/
Headers headers();
- /**
- * Return the non-deserialized byte[] of the input message key if the context has been triggered by a message.
- *
- * <p> If this method is invoked within a {@link Punctuator#punctuate(long)
- * punctuation callback}, or while processing a record that was forwarded by a punctuation
- * callback, it will return {@code null}.
- *
- * <p> If this method is invoked in a sub-topology due to a repartition, the returned key would be one sent
- * to the repartition topic.
- *
- * @return the raw byte of the key of the source message
- */
- byte[] sourceRawKey();
-
- /**
- * Return the non-deserialized byte[] of the input message value if the context has been triggered by a message.
- *
- * <p> If this method is invoked within a {@link Punctuator#punctuate(long)
- * punctuation callback}, or while processing a record that was forwarded by a punctuation
- * callback, it will return {@code null}.
- *
- * <p> If this method is invoked in a sub-topology due to a repartition, the returned value would be one sent
- * to the repartition topic.
- *
- * @return the raw byte of the value of the source message
- */
- byte[] sourceRawValue();
-
/**
* Return the current processor node ID.
*
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 6ae0170bfc9..25aa00f7a92 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
@@ -30,9 +30,28 @@ public interface ProductionExceptionHandler extends Configurable {
*
* @param record The record that failed to produce
* @param exception The exception that occurred during production
+ * @deprecated Since 3.9. Use {@link #handle(ErrorHandlerContext, ProducerRecord, Exception)} instead.
*/
- ProductionExceptionHandlerResponse handle(final ProducerRecord<byte[], byte[]> record,
- final Exception exception);
+ @Deprecated
+ default ProductionExceptionHandlerResponse handle(final ProducerRecord<byte[], byte[]> record,
+ final Exception exception) {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * Inspect a record that we attempted to produce, and the exception that resulted
+ * from attempting to produce it and determine whether or not to continue processing.
+ *
+ * @param context The error handler context metadata
+ * @param record The record that failed to produce
+ * @param exception The exception that occurred during production
+ */
+ @SuppressWarnings("deprecation")
+ default ProductionExceptionHandlerResponse handle(final ErrorHandlerContext context,
+ final ProducerRecord<byte[], byte[]> record,
+ final Exception exception) {
+ return handle(record, exception);
+ }
/**
* Handles serialization exception and determine if the process should continue. The default implementation is to
@@ -40,12 +59,31 @@ 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
default ProductionExceptionHandlerResponse handleSerializationException(final ProducerRecord record,
final Exception exception) {
return ProductionExceptionHandlerResponse.FAIL;
}
+ /**
+ * Handles serialization exception and determine if the process should continue. The default implementation is to
+ * fail the process.
+ *
+ * @param context the error handler context metadata
+ * @param record the record that failed to serialize
+ * @param exception the exception that occurred during serialization
+ * @param origin the origin of the serialization exception
+ */
+ @SuppressWarnings("deprecation")
+ default ProductionExceptionHandlerResponse handleSerializationException(final ErrorHandlerContext context,
+ final ProducerRecord record,
+ final Exception exception,
+ final SerializationExceptionOrigin origin) {
+ return handleSerializationException(record, exception);
+ }
+
enum ProductionExceptionHandlerResponse {
/* continue processing */
CONTINUE(0, "CONTINUE"),
@@ -68,4 +106,11 @@ public interface ProductionExceptionHandler extends Configurable {
this.name = name;
}
}
+
+ enum SerializationExceptionOrigin {
+ /* serialization exception occurred during serialization of the key */
+ KEY,
+ /* serialization exception occurred during serialization of the value */
+ VALUE
+ }
}
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 ff79860d77e..aa066fb6da9 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
@@ -28,8 +28,6 @@ public class DefaultErrorHandlerContext implements ErrorHandlerContext {
private final int partition;
private final long offset;
private final Headers headers;
- private final byte[] sourceRawKey;
- private final byte[] sourceRawValue;
private final String processorNodeId;
private final TaskId taskId;
@@ -37,16 +35,12 @@ public class DefaultErrorHandlerContext implements ErrorHandlerContext {
final int partition,
final long offset,
final Headers headers,
- final byte[] sourceRawKey,
- final byte[] sourceRawValue,
final String processorNodeId,
final TaskId taskId) {
this.topic = topic;
this.partition = partition;
this.offset = offset;
this.headers = headers;
- this.sourceRawKey = sourceRawKey;
- this.sourceRawValue = sourceRawValue;
this.processorNodeId = processorNodeId;
this.taskId = taskId;
}
@@ -71,16 +65,6 @@ public class DefaultErrorHandlerContext implements ErrorHandlerContext {
return headers;
}
- @Override
- public byte[] sourceRawKey() {
- return sourceRawKey;
- }
-
- @Override
- public byte[] sourceRawValue() {
- return sourceRawValue;
- }
-
@Override
public String processorNodeId() {
return processorNodeId;
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/CorruptedRecord.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/CorruptedRecord.java
index 1bc8cb51092..d31a29883ca 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/CorruptedRecord.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/CorruptedRecord.java
@@ -28,7 +28,7 @@ import java.util.Objects;
public class CorruptedRecord extends StampedRecord {
CorruptedRecord(final ConsumerRecord<byte[], byte[]> rawRecord) {
- super(rawRecord, ConsumerRecord.NO_TIMESTAMP, rawRecord);
+ super(rawRecord, ConsumerRecord.NO_TIMESTAMP);
}
@Override
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 b1263ddc58d..6b7214a9ed1 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
@@ -307,8 +307,7 @@ public class GlobalStateManagerImpl implements GlobalStateManager {
record.offset(),
record.partition(),
record.topic(),
- record.headers(),
- record);
+ record.headers());
globalProcessorContext.setRecordContext(recordContext);
try {
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java
index 1713efb52a9..12a6beedbcd 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java
@@ -113,8 +113,7 @@ public class GlobalStateUpdateTask implements GlobalStateMaintainer {
deserialized.offset(),
deserialized.partition(),
deserialized.topic(),
- deserialized.headers(),
- record);
+ deserialized.headers());
processorContext.setRecordContext(recordContext);
processorContext.setCurrentNode(sourceNodeAndDeserializer.sourceNode());
final Record<Object, Object> toProcess = new Record<>(
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorAdapter.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorAdapter.java
index a5d88f5a7f8..79db3847cfb 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorAdapter.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorAdapter.java
@@ -66,8 +66,7 @@ public final class ProcessorAdapter<KIn, VIn, KOut, VOut> implements Processor<K
context.offset(),
context.partition(),
context.topic(),
- record.headers(),
- processorRecordContext.rawRecord()
+ record.headers()
));
delegate.process(record.key(), record.value());
} finally {
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java
index 6a79434b622..b484d26f0fe 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java
@@ -261,8 +261,7 @@ public class ProcessorContextImpl extends AbstractProcessorContext<Object, Objec
recordContext.offset(),
recordContext.partition(),
recordContext.topic(),
- record.headers(),
- recordContext.rawRecord());
+ record.headers());
}
if (childName == 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 eabf9e3d5c4..c30d42ba746 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
@@ -209,8 +209,6 @@ public class ProcessorNode<KIn, VIn, KOut, VOut> {
internalProcessorContext.partition(),
internalProcessorContext.offset(),
internalProcessorContext.headers(),
- internalProcessorContext.recordContext().rawRecord().key(),
- internalProcessorContext.recordContext().rawRecord().value(),
internalProcessorContext.currentNode().name(),
internalProcessorContext.taskId());
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorRecordContext.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorRecordContext.java
index 3d1ce0529e6..839baaad875 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorRecordContext.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorRecordContext.java
@@ -16,7 +16,6 @@
*/
package org.apache.kafka.streams.processor.internals;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeader;
@@ -38,28 +37,17 @@ public class ProcessorRecordContext implements RecordContext, RecordMetadata {
private final String topic;
private final int partition;
private final Headers headers;
- private final ConsumerRecord<byte[], byte[]> rawRecord;
public ProcessorRecordContext(final long timestamp,
final long offset,
final int partition,
final String topic,
final Headers headers) {
- this(timestamp, offset, partition, topic, headers, null);
- }
-
- public ProcessorRecordContext(final long timestamp,
- final long offset,
- final int partition,
- final String topic,
- final Headers headers,
- final ConsumerRecord<byte[], byte[]> rawRecord) {
this.timestamp = timestamp;
this.offset = offset;
this.topic = topic;
this.partition = partition;
this.headers = Objects.requireNonNull(headers);
- this.rawRecord = rawRecord;
}
@Override
@@ -87,10 +75,6 @@ public class ProcessorRecordContext implements RecordContext, RecordMetadata {
return headers;
}
- public ConsumerRecord<byte[], byte[]> rawRecord() {
- return rawRecord;
- }
-
public long residentMemorySizeEstimate() {
long size = 0;
size += Long.BYTES; // value.context.timestamp
@@ -189,7 +173,7 @@ public class ProcessorRecordContext implements RecordContext, RecordMetadata {
headers = new RecordHeaders(headerArr);
}
- return new ProcessorRecordContext(timestamp, offset, partition, topic, headers, null);
+ return new ProcessorRecordContext(timestamp, offset, partition, topic, headers);
}
@Override
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 16d67666ccb..35097153a5f 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
@@ -44,6 +44,7 @@ import org.apache.kafka.streams.errors.ProductionExceptionHandler.ProductionExce
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.processor.StreamPartitioner;
import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
@@ -198,57 +199,47 @@ public class RecordCollectorImpl implements RecordCollector {
final byte[] valBytes;
try {
keyBytes = keySerializer.serialize(topic, headers, key);
- valBytes = valueSerializer.serialize(topic, headers, value);
} catch (final ClassCastException exception) {
- final String keyClass = key == null ? "unknown because key is null" : key.getClass().getName();
- final String valueClass = value == null ? "unknown because value is null" : value.getClass().getName();
- throw new StreamsException(
- String.format(
- "ClassCastException while producing data to topic %s. " +
- "A serializer (key: %s / value: %s) is not compatible to the actual key or value type " +
- "(key type: %s / value type: %s). " +
- "Change the default Serdes in StreamConfig or provide correct Serdes via method parameters " +
- "(for example if using the DSL, `#to(String topic, Produced<K, V> produced)` with " +
- "`Produced.keySerde(WindowedSerdes.timeWindowedSerdeFrom(String.class))`).",
- topic,
- keySerializer.getClass().getName(),
- valueSerializer.getClass().getName(),
- keyClass,
- valueClass),
+ throw createStreamsExceptionForKeyClassCastException(
+ topic,
+ key,
+ keySerializer,
exception);
} catch (final Exception exception) {
- final ProducerRecord<K, V> record = new ProducerRecord<>(topic, partition, timestamp, key, value, headers);
- final ProductionExceptionHandler.ProductionExceptionHandlerResponse response;
-
- log.debug(String.format("Error serializing record to topic %s", topic), exception);
-
- try {
- response = productionExceptionHandler.handleSerializationException(record, exception);
- } catch (final Exception e) {
- log.error("Fatal when handling serialization exception", e);
- recordSendError(topic, e, null);
- return;
- }
-
- if (response == ProductionExceptionHandlerResponse.FAIL) {
- throw new StreamsException(
- String.format(
- "Unable to serialize record. ProducerRecord(topic=[%s], partition=[%d], timestamp=[%d]",
- topic,
- partition,
- timestamp),
- exception
- );
- }
-
- log.warn("Unable to serialize record, continue processing. " +
- "ProducerRecord(topic=[{}], partition=[{}], timestamp=[{}])",
- topic,
- partition,
- timestamp);
-
- droppedRecordsSensor.record();
+ handleException(
+ ProductionExceptionHandler.SerializationExceptionOrigin.KEY,
+ topic,
+ key,
+ value,
+ headers,
+ partition,
+ timestamp,
+ processorNodeId,
+ context,
+ exception);
+ return;
+ }
+ try {
+ valBytes = valueSerializer.serialize(topic, headers, value);
+ } catch (final ClassCastException exception) {
+ throw createStreamsExceptionForValueClassCastException(
+ topic,
+ value,
+ valueSerializer,
+ exception);
+ } catch (final Exception exception) {
+ handleException(
+ ProductionExceptionHandler.SerializationExceptionOrigin.VALUE,
+ topic,
+ key,
+ value,
+ headers,
+ partition,
+ timestamp,
+ processorNodeId,
+ context,
+ exception);
return;
}
@@ -285,7 +276,7 @@ public class RecordCollectorImpl implements RecordCollector {
topicProducedSensor.record(bytesProduced, context.currentSystemTimeMs());
}
} else {
- recordSendError(topic, exception, serializedRecord);
+ recordSendError(topic, exception, serializedRecord, context, processorNodeId);
// KAFKA-7510 only put message key and value in TRACE level log so we don't leak data by default
log.trace("Failed record: (key {} value {} timestamp {}) topic=[{}] partition=[{}]", key, value, timestamp, topic, partition);
@@ -293,7 +284,96 @@ public class RecordCollectorImpl implements RecordCollector {
});
}
- private void recordSendError(final String topic, final Exception exception, final ProducerRecord<byte[], byte[]> serializedRecord) {
+ private <K, V> void handleException(final ProductionExceptionHandler.SerializationExceptionOrigin origin,
+ final String topic,
+ final K key,
+ final V value,
+ final Headers headers,
+ final Integer partition,
+ final Long timestamp,
+ final String processorNodeId,
+ final InternalProcessorContext<Void, Void> context,
+ final Exception exception) {
+ 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);
+
+ try {
+ final DefaultErrorHandlerContext errorHandlerContext = new DefaultErrorHandlerContext(
+ context.recordContext().topic(),
+ context.recordContext().partition(),
+ context.recordContext().offset(),
+ context.recordContext().headers(),
+ processorNodeId,
+ taskId);
+ 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;
+ }
+
+ if (response == ProductionExceptionHandlerResponse.FAIL) {
+ throw new StreamsException(
+ String.format(
+ "Unable to serialize record. ProducerRecord(topic=[%s], partition=[%d], timestamp=[%d]",
+ topic,
+ partition,
+ timestamp),
+ exception
+ );
+ }
+
+ log.warn("Unable to serialize record, continue processing. " +
+ "ProducerRecord(topic=[{}], partition=[{}], timestamp=[{}])",
+ topic,
+ partition,
+ timestamp);
+
+ droppedRecordsSensor.record();
+ }
+ private <K> StreamsException createStreamsExceptionForKeyClassCastException(final String topic,
+ final K key,
+ final Serializer<K> keySerializer,
+ final ClassCastException exception) {
+ final String keyClass = key == null ? "unknown because key is null" : key.getClass().getName();
+ return new StreamsException(
+ String.format(
+ "ClassCastException while producing data to topic %s. " +
+ "The key serializer %s is not compatible to the actual key type: %s. " +
+ "Change the default key serde in StreamConfig or provide the correct key serde via method parameters " +
+ "(for example if using the DSL, `#to(String topic, Produced<K, V> produced)` with " +
+ "`Produced.keySerde(WindowedSerdes.timeWindowedSerdeFrom(String.class))`).",
+ topic,
+ keySerializer.getClass().getName(),
+ keyClass),
+ exception);
+ }
+
+ private <V> StreamsException createStreamsExceptionForValueClassCastException(final String topic,
+ final V value,
+ final Serializer<V> valueSerializer,
+ final ClassCastException exception) {
+ final String valueClass = value == null ? "unknown because value is null" : value.getClass().getName();
+ return new StreamsException(
+ String.format(
+ "ClassCastException while producing data to topic %s. " +
+ "The value serializer %s is not compatible to the actual value type: %s. " +
+ "Change the default value serde in StreamConfig or provide the correct value serde via method parameters " +
+ "(for example if using the DSL, `#to(String topic, Produced<K, V> produced)` with " +
+ "`Produced.valueSerde(WindowedSerdes.timeWindowedSerdeFrom(String.class))`).",
+ topic,
+ valueSerializer.getClass().getName(),
+ valueClass),
+ exception);
+ }
+
+ private void recordSendError(final String topic,
+ final Exception exception,
+ final ProducerRecord<byte[], byte[]> serializedRecord,
+ final InternalProcessorContext<Void, Void> context,
+ final String processorNodeId) {
String errorMessage = String.format(SEND_EXCEPTION_MESSAGE, topic, taskId, exception.toString());
if (isFatalException(exception)) {
@@ -314,7 +394,15 @@ public class RecordCollectorImpl implements RecordCollector {
"`delivery.timeout.ms` to a larger value to wait longer for such scenarios and avoid timeout errors";
sendException.set(new TaskCorruptedException(Collections.singleton(taskId)));
} else {
- if (productionExceptionHandler.handle(serializedRecord, exception) == ProductionExceptionHandlerResponse.FAIL) {
+ final DefaultErrorHandlerContext errorHandlerContext = new DefaultErrorHandlerContext(
+ context.recordContext().topic(),
+ context.recordContext().partition(),
+ context.recordContext().offset(),
+ context.recordContext().headers(),
+ processorNodeId,
+ taskId);
+
+ if (productionExceptionHandler.handle(errorHandlerContext, serializedRecord, exception) == ProductionExceptionHandlerResponse.FAIL) {
errorMessage += "\nException handler choose to FAIL the processing, no more records would be sent.";
sendException.set(new StreamsException(errorMessage, exception));
} else {
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordQueue.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordQueue.java
index a3c9ea67f06..a6b30a07ef9 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordQueue.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordQueue.java
@@ -230,7 +230,7 @@ public class RecordQueue {
droppedRecordsSensor.record();
continue;
}
- headRecord = new StampedRecord(deserialized, timestamp, raw);
+ headRecord = new StampedRecord(deserialized, timestamp);
headRecordSizeInBytes = consumerRecordSizeInBytes(raw);
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java
index 871cb2284ee..6e79616d30a 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java
@@ -88,8 +88,7 @@ public class SinkNode<KIn, VIn> extends ProcessorNode<KIn, VIn, Void, Void> {
context.offset(),
context.partition(),
context.topic(),
- record.headers(),
- context.recordContext().rawRecord()
+ record.headers()
);
final String topic = topicExtractor.extract(key, value, contextForExtraction);
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StampedRecord.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StampedRecord.java
index d82cd98ed7e..71e3ca2e3ce 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StampedRecord.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StampedRecord.java
@@ -20,11 +20,9 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Headers;
public class StampedRecord extends Stamped<ConsumerRecord<?, ?>> {
- private final ConsumerRecord<byte[], byte[]> rawRecord;
- public StampedRecord(final ConsumerRecord<?, ?> record, final long timestamp, final ConsumerRecord<byte[], byte[]> rawRecord) {
+ public StampedRecord(final ConsumerRecord<?, ?> record, final long timestamp) {
super(record, timestamp);
- this.rawRecord = rawRecord;
}
public String topic() {
@@ -51,20 +49,6 @@ public class StampedRecord extends Stamped<ConsumerRecord<?, ?>> {
return value.headers();
}
- public ConsumerRecord<byte[], byte[]> rawRecord() {
- return rawRecord;
- }
-
- @Override
- public boolean equals(final Object other) {
- return super.equals(other);
- }
-
- @Override
- public int hashCode() {
- return super.hashCode();
- }
-
@Override
public String toString() {
return value.toString() + ", timestamp = " + timestamp;
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 30b9038aa6a..8cbe5780b90 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
@@ -844,8 +844,7 @@ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator,
record.offset(),
record.partition(),
record.topic(),
- record.headers(),
- record.rawRecord()
+ record.headers()
);
updateProcessorContext(currNode, wallClockTime, recordContext);
@@ -906,8 +905,7 @@ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator,
-1L,
-1,
null,
- new RecordHeaders(),
- null
+ new RecordHeaders()
);
updateProcessorContext(node, time.milliseconds(), recordContext);
diff --git a/streams/src/test/java/org/apache/kafka/streams/errors/AlwaysContinueProductionExceptionHandler.java b/streams/src/test/java/org/apache/kafka/streams/errors/AlwaysContinueProductionExceptionHandler.java
deleted file mode 100644
index be1e98e4a71..00000000000
--- a/streams/src/test/java/org/apache/kafka/streams/errors/AlwaysContinueProductionExceptionHandler.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.kafka.streams.errors;
-
-import org.apache.kafka.clients.producer.ProducerRecord;
-
-import java.util.Map;
-
-/**
- * Production exception handler that always instructs streams to continue when an exception
- * happens while attempting to produce result records.
- */
-public class AlwaysContinueProductionExceptionHandler implements ProductionExceptionHandler {
- @Override
- public ProductionExceptionHandlerResponse handle(final ProducerRecord<byte[], byte[]> record,
- final Exception exception) {
- return ProductionExceptionHandlerResponse.CONTINUE;
- }
-
- @Override
- public ProductionExceptionHandlerResponse handleSerializationException(final ProducerRecord record,
- final Exception exception) {
- return ProductionExceptionHandlerResponse.CONTINUE;
- }
-
- @Override
- public void configure(final Map<String, ?> configs) {
- // ignore
- }
-}
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 bb29ac64f02..6c1a64344ec 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
@@ -176,8 +176,6 @@ public class ProcessingExceptionHandlerIntegrationTest {
}
private static void assertProcessingExceptionHandlerInputs(final ErrorHandlerContext context, final Record<?, ?> record, final Exception exception) {
- assertTrue(Arrays.asList("ID123-2-ERR", "ID123-5-ERR").contains(new String(context.sourceRawKey())));
- assertTrue(Arrays.asList("ID123-A2", "ID123-A5").contains(new String(context.sourceRawValue())));
assertTrue(Arrays.asList("ID123-2-ERR", "ID123-5-ERR").contains((String) record.key()));
assertTrue(Arrays.asList("ID123-A2", "ID123-A5").contains((String) record.value()));
assertEquals("TOPIC_NAME", context.topic());
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java
index 6be033cb2dc..df3f9276863 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java
@@ -16,7 +16,6 @@
*/
package org.apache.kafka.streams.processor.internals;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.InvalidOffsetException;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.header.internals.RecordHeaders;
@@ -58,7 +57,6 @@ import java.util.Properties;
import java.util.Set;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.ROLLUP_VALUE;
-import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -315,8 +313,7 @@ public class ProcessorNodeTest {
OFFSET,
PARTITION,
TOPIC,
- new RecordHeaders(),
- new ConsumerRecord<>(TOPIC, PARTITION, OFFSET, KEY.getBytes(), VALUE.getBytes())));
+ new RecordHeaders()));
when(internalProcessorContext.currentNode()).thenReturn(new ProcessorNode<>(NAME));
return internalProcessorContext;
@@ -337,8 +334,6 @@ public class ProcessorNodeTest {
assertEquals(internalProcessorContext.topic(), context.topic());
assertEquals(internalProcessorContext.partition(), context.partition());
assertEquals(internalProcessorContext.offset(), context.offset());
- assertArrayEquals(internalProcessorContext.recordContext().rawRecord().key(), context.sourceRawKey());
- assertArrayEquals(internalProcessorContext.recordContext().rawRecord().value(), context.sourceRawValue());
assertEquals(internalProcessorContext.currentNode().name(), context.processorNodeId());
assertEquals(internalProcessorContext.taskId(), context.taskId());
assertEquals(KEY, record.key());
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 5222759bc5e..26231028851 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
@@ -48,8 +48,8 @@ import org.apache.kafka.common.utils.LogCaptureAppender;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.streams.StreamsConfig;
-import org.apache.kafka.streams.errors.AlwaysContinueProductionExceptionHandler;
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.StreamsException;
import org.apache.kafka.streams.errors.TaskMigratedException;
@@ -96,6 +96,7 @@ import static org.hamcrest.core.IsInstanceOf.instanceOf;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
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.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
@@ -918,9 +919,9 @@ public class RecordCollectorTest {
expected.getMessage(),
equalTo(
"ClassCastException while producing data to topic topic. " +
- "A serializer (key: org.apache.kafka.common.serialization.LongSerializer / value: org.apache.kafka.common.serialization.StringSerializer) " +
- "is not compatible to the actual key or value type (key type: java.lang.String / value type: java.lang.String). " +
- "Change the default Serdes in StreamConfig or provide correct Serdes via method parameters " +
+ "The key serializer org.apache.kafka.common.serialization.LongSerializer " +
+ "is not compatible to the actual key type: java.lang.String. " +
+ "Change the default key serde in StreamConfig or provide the correct key serde via method parameters " +
"(for example if using the DSL, `#to(String topic, Produced<K, V> produced)` with `Produced.keySerde(WindowedSerdes.timeWindowedSerdeFrom(String.class))`).")
);
}
@@ -946,9 +947,9 @@ public class RecordCollectorTest {
expected.getMessage(),
equalTo(
"ClassCastException while producing data to topic topic. " +
- "A serializer (key: org.apache.kafka.common.serialization.LongSerializer / value: org.apache.kafka.common.serialization.StringSerializer) " +
- "is not compatible to the actual key or value type (key type: java.lang.String / value type: unknown because value is null). " +
- "Change the default Serdes in StreamConfig or provide correct Serdes via method parameters " +
+ "The key serializer org.apache.kafka.common.serialization.LongSerializer " +
+ "is not compatible to the actual key type: java.lang.String. " +
+ "Change the default key serde in StreamConfig or provide the correct key serde via method parameters " +
"(for example if using the DSL, `#to(String topic, Produced<K, V> produced)` with `Produced.keySerde(WindowedSerdes.timeWindowedSerdeFrom(String.class))`).")
);
}
@@ -974,10 +975,10 @@ public class RecordCollectorTest {
expected.getMessage(),
equalTo(
"ClassCastException while producing data to topic topic. " +
- "A serializer (key: org.apache.kafka.common.serialization.StringSerializer / value: org.apache.kafka.common.serialization.LongSerializer) " +
- "is not compatible to the actual key or value type (key type: java.lang.String / value type: java.lang.String). " +
- "Change the default Serdes in StreamConfig or provide correct Serdes via method parameters " +
- "(for example if using the DSL, `#to(String topic, Produced<K, V> produced)` with `Produced.keySerde(WindowedSerdes.timeWindowedSerdeFrom(String.class))`).")
+ "The value serializer org.apache.kafka.common.serialization.LongSerializer " +
+ "is not compatible to the actual value type: java.lang.String. " +
+ "Change the default value serde in StreamConfig or provide the correct value serde via method parameters " +
+ "(for example if using the DSL, `#to(String topic, Produced<K, V> produced)` with `Produced.valueSerde(WindowedSerdes.timeWindowedSerdeFrom(String.class))`).")
);
}
@@ -1002,10 +1003,10 @@ public class RecordCollectorTest {
expected.getMessage(),
equalTo(
"ClassCastException while producing data to topic topic. " +
- "A serializer (key: org.apache.kafka.common.serialization.StringSerializer / value: org.apache.kafka.common.serialization.LongSerializer) " +
- "is not compatible to the actual key or value type (key type: unknown because key is null / value type: java.lang.String). " +
- "Change the default Serdes in StreamConfig or provide correct Serdes via method parameters " +
- "(for example if using the DSL, `#to(String topic, Produced<K, V> produced)` with `Produced.keySerde(WindowedSerdes.timeWindowedSerdeFrom(String.class))`).")
+ "The value serializer org.apache.kafka.common.serialization.LongSerializer " +
+ "is not compatible to the actual value type: java.lang.String. " +
+ "Change the default value serde in StreamConfig or provide the correct value serde via method parameters " +
+ "(for example if using the DSL, `#to(String topic, Produced<K, V> produced)` with `Produced.valueSerde(WindowedSerdes.timeWindowedSerdeFrom(String.class))`).")
);
}
@@ -1171,11 +1172,11 @@ public class RecordCollectorTest {
topology
);
- collector.send(topic, "3", "0", null, null, stringSerializer, stringSerializer, null, null, streamPartitioner);
+ collector.send(topic, "3", "0", null, null, stringSerializer, stringSerializer, sinkNodeName, context, streamPartitioner);
final StreamsException thrown = assertThrows(
StreamsException.class,
- () -> collector.send(topic, "3", "0", null, null, stringSerializer, stringSerializer, null, null, streamPartitioner)
+ () -> collector.send(topic, "3", "0", null, null, stringSerializer, stringSerializer, sinkNodeName, context, streamPartitioner)
);
assertEquals(exception, thrown.getCause());
assertThat(
@@ -1198,7 +1199,7 @@ public class RecordCollectorTest {
topology
);
- collector.send(topic, "3", "0", null, null, stringSerializer, stringSerializer, null, null, streamPartitioner);
+ collector.send(topic, "3", "0", null, null, stringSerializer, stringSerializer, sinkNodeName, context, streamPartitioner);
final StreamsException thrown = assertThrows(StreamsException.class, collector::flush);
assertEquals(exception, thrown.getCause());
@@ -1222,7 +1223,7 @@ public class RecordCollectorTest {
topology
);
- collector.send(topic, "3", "0", null, null, stringSerializer, stringSerializer, null, null, streamPartitioner);
+ collector.send(topic, "3", "0", null, null, stringSerializer, stringSerializer, sinkNodeName, context, streamPartitioner);
final StreamsException thrown = assertThrows(StreamsException.class, collector::closeClean);
assertEquals(exception, thrown.getCause());
@@ -1241,7 +1242,7 @@ public class RecordCollectorTest {
logContext,
taskId,
getExceptionalStreamsProducerOnSend(exception),
- new AlwaysContinueProductionExceptionHandler(),
+ new ProductionExceptionHandlerMock(ProductionExceptionHandler.ProductionExceptionHandlerResponse.CONTINUE),
streamsMetrics,
topology
);
@@ -1268,7 +1269,7 @@ public class RecordCollectorTest {
logContext,
taskId,
getExceptionalStreamsProducerOnSend(exception),
- new AlwaysContinueProductionExceptionHandler(),
+ new ProductionExceptionHandlerMock(ProductionExceptionHandler.ProductionExceptionHandlerResponse.CONTINUE),
streamsMetrics,
topology
);
@@ -1292,7 +1293,7 @@ public class RecordCollectorTest {
logContext,
taskId,
getExceptionalStreamsProducerOnSend(exception),
- new AlwaysContinueProductionExceptionHandler(),
+ new ProductionExceptionHandlerMock(ProductionExceptionHandler.ProductionExceptionHandlerResponse.CONTINUE),
streamsMetrics,
topology
);
@@ -1314,8 +1315,13 @@ public class RecordCollectorTest {
final RecordCollector collector = new RecordCollectorImpl(
logContext,
taskId,
- getExceptionalStreamsProducerOnSend(new Exception()),
- new AlwaysContinueProductionExceptionHandler(),
+ getExceptionalStreamsProducerOnSend(new RuntimeException("KABOOM!")),
+ new ProductionExceptionHandlerMock(
+ ProductionExceptionHandler.ProductionExceptionHandlerResponse.CONTINUE,
+ context,
+ sinkNodeName,
+ taskId
+ ),
streamsMetrics,
topology
);
@@ -1324,7 +1330,7 @@ public class RecordCollectorTest {
LogCaptureAppender.createAndRegister(RecordCollectorImpl.class)) {
logCaptureAppender.setThreshold(Level.INFO);
- collector.send(topic, "3", "0", null, null, stringSerializer, stringSerializer, null, null, streamPartitioner);
+ collector.send(topic, "3", "0", null, null, stringSerializer, stringSerializer, sinkNodeName, context, streamPartitioner);
collector.flush();
final List<String> messages = logCaptureAppender.getMessages();
@@ -1348,7 +1354,7 @@ public class RecordCollectorTest {
));
assertEquals(1.0, metric.metricValue());
- collector.send(topic, "3", "0", null, null, stringSerializer, stringSerializer, null, null, streamPartitioner);
+ collector.send(topic, "3", "0", null, null, stringSerializer, stringSerializer, sinkNodeName, context, streamPartitioner);
collector.flush();
collector.closeClean();
}
@@ -1365,7 +1371,7 @@ public class RecordCollectorTest {
topology
);
- collector.send(topic, "3", "0", null, null, stringSerializer, stringSerializer, null, null, streamPartitioner);
+ collector.send(topic, "3", "0", null, null, stringSerializer, stringSerializer, sinkNodeName, context, streamPartitioner);
// With default handler which returns FAIL, flush() throws StreamsException with TimeoutException cause,
// otherwise it would throw a TaskCorruptedException with null cause
@@ -1386,12 +1392,17 @@ public class RecordCollectorTest {
logContext,
taskId,
getExceptionalStreamsProducerOnSend(exception),
- new AlwaysContinueProductionExceptionHandler(),
+ new ProductionExceptionHandlerMock(
+ ProductionExceptionHandler.ProductionExceptionHandlerResponse.CONTINUE,
+ context,
+ sinkNodeName,
+ taskId
+ ),
streamsMetrics,
topology
);
- collector.send(topic, "3", "0", null, null, stringSerializer, stringSerializer, null, null, streamPartitioner);
+ collector.send(topic, "3", "0", null, null, stringSerializer, stringSerializer, sinkNodeName, context, streamPartitioner);
assertDoesNotThrow(collector::flush);
}
@@ -1527,7 +1538,13 @@ public class RecordCollectorTest {
@Test
public void shouldDropRecordExceptionUsingAlwaysContinueExceptionHandler() {
try (final ErrorStringSerializer errorSerializer = new ErrorStringSerializer()) {
- final RecordCollector collector = newRecordCollector(new AlwaysContinueProductionExceptionHandler());
+ final RecordCollector collector = newRecordCollector(new ProductionExceptionHandlerMock(
+ ProductionExceptionHandler.ProductionExceptionHandlerResponse.CONTINUE,
+ context,
+ sinkNodeName,
+ taskId,
+ ProductionExceptionHandler.SerializationExceptionOrigin.KEY
+ ));
collector.initialize();
collector.send(topic, "key", "val", null, 0, null, errorSerializer, stringSerializer, sinkNodeName, context);
@@ -1547,11 +1564,53 @@ public class RecordCollectorTest {
}
}
+ @Test
+ public void shouldThrowStreamsExceptionWhenValueSerializationFailedAndProductionExceptionHandlerRepliesWithFail() {
+ try (final ErrorStringSerializer errorSerializer = new ErrorStringSerializer()) {
+ final RecordCollector collector = newRecordCollector(new ProductionExceptionHandlerMock(
+ ProductionExceptionHandler.ProductionExceptionHandlerResponse.FAIL,
+ context,
+ sinkNodeName,
+ taskId,
+ ProductionExceptionHandler.SerializationExceptionOrigin.VALUE
+ ));
+ collector.initialize();
+
+ final StreamsException exception = assertThrows(
+ StreamsException.class,
+ () -> collector.send(topic, "key", "val", null, 0, null, stringSerializer, errorSerializer, sinkNodeName, context));
+
+ assertInstanceOf(RuntimeException.class, exception.getCause());
+ assertEquals("KABOOM!", exception.getCause().getMessage());
+ }
+ }
+
+ @Test
+ public void shouldThrowStreamsExceptionWhenKeySerializationFailedAndProductionExceptionHandlerRepliesWithFail() {
+ try (final ErrorStringSerializer errorSerializer = new ErrorStringSerializer()) {
+ final RecordCollector collector = newRecordCollector(new ProductionExceptionHandlerMock(
+ ProductionExceptionHandler.ProductionExceptionHandlerResponse.FAIL,
+ context,
+ sinkNodeName,
+ taskId,
+ ProductionExceptionHandler.SerializationExceptionOrigin.KEY
+ ));
+ collector.initialize();
+
+ final StreamsException exception = assertThrows(
+ StreamsException.class,
+ () -> collector.send(topic, "key", "val", null, 0, null, errorSerializer, stringSerializer, sinkNodeName, context));
+
+ assertInstanceOf(RuntimeException.class, exception.getCause());
+ assertEquals("KABOOM!", exception.getCause().getMessage());
+ }
+ }
+
@SuppressWarnings({"unchecked", "rawtypes"})
@Test
public void shouldNotCallProductionExceptionHandlerOnClassCastException() {
try (final ErrorStringSerializer errorSerializer = new ErrorStringSerializer()) {
- final RecordCollector collector = newRecordCollector(new AlwaysContinueProductionExceptionHandler());
+ final RecordCollector collector = newRecordCollector(new ProductionExceptionHandlerMock(ProductionExceptionHandler.ProductionExceptionHandlerResponse.CONTINUE));
collector.initialize();
assertThat(mockProducer.history().isEmpty(), equalTo(true));
@@ -1637,7 +1696,7 @@ public class RecordCollectorTest {
@Override
public byte[] serialize(final String topic, final Headers headers, final String data) {
- throw new SerializationException("Not Supported");
+ throw new SerializationException("KABOOM!");
}
}
@@ -1705,4 +1764,68 @@ public class RecordCollectorTest {
return serialize(topic, data);
}
}
+
+ public static class ProductionExceptionHandlerMock implements ProductionExceptionHandler {
+ private final ProductionExceptionHandlerResponse response;
+ private InternalProcessorContext<Void, Void> expectedContext;
+ private String expectedProcessorNodeId;
+ private TaskId expectedTaskId;
+ private SerializationExceptionOrigin expectedSerializationExceptionOrigin;
+
+ public ProductionExceptionHandlerMock(final ProductionExceptionHandlerResponse response) {
+ this.response = response;
+ }
+
+ public ProductionExceptionHandlerMock(final ProductionExceptionHandlerResponse response,
+ final InternalProcessorContext<Void, Void> context,
+ final String processorNodeId,
+ final TaskId taskId) {
+ this(response);
+ this.expectedContext = context;
+ this.expectedProcessorNodeId = processorNodeId;
+ this.expectedTaskId = taskId;
+ }
+
+ public ProductionExceptionHandlerMock(final ProductionExceptionHandlerResponse response,
+ final InternalProcessorContext<Void, Void> context,
+ final String processorNodeId,
+ final TaskId taskId,
+ final SerializationExceptionOrigin origin) {
+ this(response, context, processorNodeId, taskId);
+ this.expectedSerializationExceptionOrigin = origin;
+ }
+
+ @Override
+ public ProductionExceptionHandlerResponse handle(final ErrorHandlerContext context,
+ final ProducerRecord<byte[], byte[]> record,
+ final Exception exception) {
+ assertInputs(context, exception);
+ return response;
+ }
+
+ @Override
+ public ProductionExceptionHandlerResponse handleSerializationException(final ErrorHandlerContext context,
+ final ProducerRecord record,
+ final Exception exception,
+ final SerializationExceptionOrigin origin) {
+ assertInputs(context, exception);
+ assertEquals(expectedSerializationExceptionOrigin, origin);
+ return response;
+ }
+
+ @Override
+ public void configure(final Map<String, ?> configs) {
+ // do nothing
+ }
+
+ private void assertInputs(final ErrorHandlerContext context, final Exception exception) {
+ assertEquals(expectedContext.recordContext().topic(), context.topic());
+ assertEquals(expectedContext.recordContext().partition(), context.partition());
+ assertEquals(expectedContext.recordContext().offset(), context.offset());
+ assertEquals(expectedProcessorNodeId, context.processorNodeId());
+ assertEquals(expectedTaskId, context.taskId());
+ assertInstanceOf(RuntimeException.class, exception);
+ assertEquals("KABOOM!", exception.getMessage());
+ }
+ }
}
diff --git a/streams/src/test/java/org/apache/kafka/test/InternalMockProcessorContext.java b/streams/src/test/java/org/apache/kafka/test/InternalMockProcessorContext.java
index 237d1c665b8..ff9080a2de1 100644
--- a/streams/src/test/java/org/apache/kafka/test/InternalMockProcessorContext.java
+++ b/streams/src/test/java/org/apache/kafka/test/InternalMockProcessorContext.java
@@ -239,6 +239,13 @@ public class InternalMockProcessorContext<KOut, VOut>
appConfigs(),
IQ_CONSISTENCY_OFFSET_VECTOR_ENABLED,
false);
+ this.recordContext = new ProcessorRecordContext(
+ 0,
+ 0,
+ 0,
+ "topic",
+ new RecordHeaders()
+ );
}
@Override
(kafka) 03/03: HOTFIX: fix compilation error
Posted by mj...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch 3.9
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit b8532070f79ff4d13f61e4cd766b9190299aab75
Author: Matthias J. Sax <ma...@confluent.io>
AuthorDate: Mon Jul 29 21:07:49 2024 -0700
HOTFIX: fix compilation error
---
.../apache/kafka/streams/processor/internals/ProcessorNode.java | 2 +-
.../kafka/streams/processor/internals/RecordCollectorImpl.java | 8 ++++++--
2 files changed, 7 insertions(+), 3 deletions(-)
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 65eec47cb1d..763edc9a045 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
@@ -205,7 +205,7 @@ public class ProcessorNode<KIn, VIn, KOut, VOut> {
throw e;
} catch (final Exception e) {
final ErrorHandlerContext errorHandlerContext = new DefaultErrorHandlerContext(
- null,
+ null, // only required to pass for DeserializationExceptionHandler
internalProcessorContext.topic(),
internalProcessorContext.partition(),
internalProcessorContext.offset(),
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 35097153a5f..de4afc2c924 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
@@ -301,12 +301,14 @@ public class RecordCollectorImpl implements RecordCollector {
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);
+ taskId
+ );
response = productionExceptionHandler.handleSerializationException(errorHandlerContext, record, exception, origin);
} catch (final Exception e) {
log.error("Fatal when handling serialization exception", e);
@@ -395,12 +397,14 @@ public class RecordCollectorImpl implements RecordCollector {
sendException.set(new TaskCorruptedException(Collections.singleton(taskId)));
} else {
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);
+ taskId
+ );
if (productionExceptionHandler.handle(errorHandlerContext, serializedRecord, exception) == ProductionExceptionHandlerResponse.FAIL) {
errorMessage += "\nException handler choose to FAIL the processing, no more records would be sent.";