Skip to content

KAFKA-16448: Handle fatal user exception during processing error #16675

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mjsax merged 5 commits into from
Jul 31, 2024
Merged

KAFKA-16448: Handle fatal user exception during processing error #16675

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
2be7af8
KAFKA-16448 Handle fatal user exception during processing error
sebastienviale Jul 24, 2024
b93561b
KAFKA-16448 Handle fatal user exception during processing error
sebastienviale Jul 30, 2024
92235e5
KAFKA-16448: Fix processing exception handler throwing exception itself
loicgreffier Jul 30, 2024
458ea81
KAFKA-16448: Fix spaces
loicgreffier Jul 30, 2024
9587069
KAFKA-16448: Fix spaces
loicgreffier Jul 30, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -213,9 +213,13 @@ public void process(final Record<KIn, VIn> record) {
internalProcessorContext.currentNode().name(),
internalProcessorContext.taskId());

final ProcessingExceptionHandler.ProcessingHandlerResponse response = processingExceptionHandler
.handle(errorHandlerContext, record, e);
final ProcessingExceptionHandler.ProcessingHandlerResponse response;

try {
response = processingExceptionHandler.handle(errorHandlerContext, record, e);
} catch (final Exception fatalUserException) {
Copy link
Copy Markdown
Member

@cadonna cadonna Jul 24, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you please also add a unit test for this?

Copy link
Copy Markdown
Contributor Author

@sebastienviale sebastienviale Jul 24, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

done

throw new FailedProcessingException(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" +
Expand Down
146 changes: 146 additions & 0 deletions .../java/org/apache/kafka/streams/integration/ProcessingExceptionHandlerIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,10 +45,12 @@
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;

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.assertIterableEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
Expand Down Expand Up @@ -149,9 +151,150 @@ public void shouldContinueWhenProcessingExceptionOccurs() {
}
}

@Test
public void shouldStopOnFailedProcessorWhenProcessingExceptionOccursInFailProcessingExceptionHandler() {
final KeyValue<String, String> event = new KeyValue<>("ID123-1", "ID123-A1");
final KeyValue<String, String> eventError = new KeyValue<>("ID123-2-ERR", "ID123-A2");

final MockProcessorSupplier<String, String, Void, Void> processor = new MockProcessorSupplier<>();
final StreamsBuilder builder = new StreamsBuilder();
final AtomicBoolean isExecuted = new AtomicBoolean(false);
builder
.stream("TOPIC_NAME", Consumed.with(Serdes.String(), Serdes.String()))
.map(KeyValue::new)
.mapValues(value -> value)
.process(runtimeErrorProcessorSupplierMock())
.map((k, v) -> {
isExecuted.set(true);
return KeyValue.pair(k, v);
})
.process(processor);

final Properties properties = new Properties();
properties.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG, FailProcessingExceptionHandlerMockTest.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());
isExecuted.set(false);
inputTopic.pipeInput(event.key, event.value, Instant.EPOCH);
assertTrue(isExecuted.get());
isExecuted.set(false);
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"));
assertFalse(isExecuted.get());
}
}

@Test
public void shouldStopOnFailedProcessorWhenProcessingExceptionOccursInContinueProcessingExceptionHandler() {
final KeyValue<String, String> event = new KeyValue<>("ID123-1", "ID123-A1");
final KeyValue<String, String> eventFalse = new KeyValue<>("ID123-2-ERR", "ID123-A2");

final MockProcessorSupplier<String, String, Void, Void> processor = new MockProcessorSupplier<>();
final StreamsBuilder builder = new StreamsBuilder();
final AtomicBoolean isExecuted = new AtomicBoolean(false);
builder
.stream("TOPIC_NAME", Consumed.with(Serdes.String(), Serdes.String()))
.map(KeyValue::new)
.mapValues(value -> value)
.process(runtimeErrorProcessorSupplierMock())
.map((k, v) -> {
isExecuted.set(true);
return KeyValue.pair(k, v);
})
.process(processor);

final Properties properties = new Properties();
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());
isExecuted.set(false);
inputTopic.pipeInput(event.key, event.value, Instant.EPOCH);
assertTrue(isExecuted.get());
isExecuted.set(false);
inputTopic.pipeInput(eventFalse.key, eventFalse.value, Instant.EPOCH);
assertFalse(isExecuted.get());
}
}

@Test
public void shouldStopProcessingWhenFatalUserExceptionInFailProcessingExceptionHandler() {
final KeyValue<String, String> event = new KeyValue<>("ID123-1", "ID123-A1");
final KeyValue<String, String> eventError = new KeyValue<>("ID123-ERR-FATAL", "ID123-A2");

final MockProcessorSupplier<String, String, Void, Void> processor = new MockProcessorSupplier<>();
final StreamsBuilder builder = new StreamsBuilder();
final AtomicBoolean isExecuted = new AtomicBoolean(false);
builder
.stream("TOPIC_NAME", Consumed.with(Serdes.String(), Serdes.String()))
.map(KeyValue::new)
.mapValues(value -> value)
.process(runtimeErrorProcessorSupplierMock())
.map((k, v) -> {
isExecuted.set(true);
return KeyValue.pair(k, v);
})
.process(processor);

final Properties properties = new Properties();
properties.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG, FailProcessingExceptionHandlerMockTest.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());
isExecuted.set(false);
inputTopic.pipeInput(event.key, event.value, Instant.EPOCH);
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());
assertFalse(isExecuted.get());
}
}

@Test
public void shouldStopProcessingWhenFatalUserExceptionInContinueProcessingExceptionHandler() {
final KeyValue<String, String> event = new KeyValue<>("ID123-1", "ID123-A1");
final KeyValue<String, String> eventError = new KeyValue<>("ID123-ERR-FATAL", "ID123-A2");

final MockProcessorSupplier<String, String, Void, Void> processor = new MockProcessorSupplier<>();
final StreamsBuilder builder = new StreamsBuilder();
final AtomicBoolean isExecuted = new AtomicBoolean(false);
builder
.stream("TOPIC_NAME", Consumed.with(Serdes.String(), Serdes.String()))
.map(KeyValue::new)
.mapValues(value -> value)
.process(runtimeErrorProcessorSupplierMock())
.map((k, v) -> {
isExecuted.set(true);
return KeyValue.pair(k, v);
})
.process(processor);

final Properties properties = new Properties();
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());
isExecuted.set(false);
inputTopic.pipeInput(event.key, event.value, Instant.EPOCH);
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());
assertFalse(isExecuted.get());
}
}

public static class ContinueProcessingExceptionHandlerMockTest 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.CONTINUE;
}
Expand All @@ -165,6 +308,9 @@ public void configure(final Map<String, ?> configs) {
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;
}
Expand Down
31 changes: 27 additions & 4 deletions streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@
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;
Expand Down Expand Up @@ -100,7 +101,7 @@ public void shouldThrowFailedProcessingExceptionWhenProcessingExceptionHandlerRe
new ProcessorNode<>(NAME, new IgnoredInternalExceptionsProcessor(), Collections.emptySet());

final InternalProcessorContext<Object, Object> internalProcessorContext = mockInternalProcessorContext();
node.init(internalProcessorContext, new ProcessingExceptionHandlerMock(ProcessingExceptionHandler.ProcessingHandlerResponse.FAIL, internalProcessorContext));
node.init(internalProcessorContext, new ProcessingExceptionHandlerMock(ProcessingExceptionHandler.ProcessingHandlerResponse.FAIL, internalProcessorContext, false));

final FailedProcessingException failedProcessingException = assertThrows(FailedProcessingException.class,
() -> node.process(new Record<>(KEY, VALUE, TIMESTAMP)));
Expand All @@ -116,7 +117,7 @@ public void shouldNotThrowFailedProcessingExceptionWhenProcessingExceptionHandle
new ProcessorNode<>(NAME, new IgnoredInternalExceptionsProcessor(), Collections.emptySet());

final InternalProcessorContext<Object, Object> internalProcessorContext = mockInternalProcessorContext();
node.init(internalProcessorContext, new ProcessingExceptionHandlerMock(ProcessingExceptionHandler.ProcessingHandlerResponse.CONTINUE, internalProcessorContext));
node.init(internalProcessorContext, new ProcessingExceptionHandlerMock(ProcessingExceptionHandler.ProcessingHandlerResponse.CONTINUE, internalProcessorContext, false));

assertDoesNotThrow(() -> node.process(new Record<>(KEY, VALUE, TIMESTAMP)));
}
Expand Down Expand Up @@ -146,6 +147,21 @@ public void shouldNotHandleInternalExceptionsThrownDuringProcessing(final String
verify(processingExceptionHandler, never()).handle(any(), any(), any());
}

@Test
public void shouldThrowFailedProcessingExceptionWhenProcessingExceptionHandlerThrowsAnException() {
final ProcessorNode<Object, Object, Object, Object> node =
new ProcessorNode<>(NAME, new IgnoredInternalExceptionsProcessor(), Collections.emptySet());

final InternalProcessorContext<Object, Object> internalProcessorContext = mockInternalProcessorContext();
node.init(internalProcessorContext, new ProcessingExceptionHandlerMock(ProcessingExceptionHandler.ProcessingHandlerResponse.CONTINUE, internalProcessorContext, true));

final FailedProcessingException failedProcessingException = assertThrows(FailedProcessingException.class,
() -> node.process(new Record<>(KEY, VALUE, TIMESTAMP)));

assertInstanceOf(RuntimeException.class, failedProcessingException.getCause());
assertEquals("KABOOM!", failedProcessingException.getCause().getMessage());
}

private static class ExceptionalProcessor implements Processor<Object, Object, Object, Object> {
@Override
public void init(final ProcessorContext<Object, Object> context) {
Expand Down Expand Up @@ -323,10 +339,14 @@ public static class ProcessingExceptionHandlerMock implements ProcessingExceptio
private final ProcessingExceptionHandler.ProcessingHandlerResponse response;
private final InternalProcessorContext<Object, Object> internalProcessorContext;

private final boolean shouldThrowException;

public ProcessingExceptionHandlerMock(final ProcessingExceptionHandler.ProcessingHandlerResponse response,
final InternalProcessorContext<Object, Object> internalProcessorContext) {
final InternalProcessorContext<Object, Object> internalProcessorContext,
final boolean shouldThrowException) {
this.response = response;
this.internalProcessorContext = internalProcessorContext;
this.shouldThrowException = shouldThrowException;
}

@Override
Expand All @@ -338,9 +358,12 @@ public ProcessingExceptionHandler.ProcessingHandlerResponse handle(final ErrorHa
assertEquals(internalProcessorContext.taskId(), context.taskId());
assertEquals(KEY, record.key());
assertEquals(VALUE, record.value());
assertTrue(exception instanceof RuntimeException);
assertInstanceOf(RuntimeException.class, exception);
assertEquals("Processing exception should be caught and handled by the processing exception handler.", exception.getMessage());

if (shouldThrowException) {
throw new RuntimeException("KABOOM!");
}
return response;
}

Expand Down