HOTFIX: fix compilation error

apache · Jul 30, 2024 · 3db4a78 · 3db4a78
1 parent faaef52
commit 3db4a78
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 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
@@ -205,7 +205,7 @@ public void process(final Record<KIn, VIn> record) {
             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
@@ -301,12 +301,14 @@ private <K, V> void handleException(final ProductionExceptionHandler.Serializati
 
         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 @@ private void recordSendError(final String topic,
                 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.";