mjsax commented on code in PR #16745:
URL: https://github.com/apache/kafka/pull/16745#discussion_r1698041422
##########
streams/src/main/java/org/apache/kafka/streams/errors/internals/DefaultErrorHandlerContext.java:
##########
@@ -81,6 +81,17 @@ public TaskId taskId() {
return taskId;
}
+ @Override
+ public String toString() {
+ return "ErrorHandlerContext{" +
+ "topic='" + topic + '\'' +
+ ", partition=" + partition +
+ ", offset=" + offset +
+ ", processorNodeId='" + processorNodeId + '\'' +
+ ", taskId=" + taskId +
+ '}';
+ }
Review Comment:
Omitting `headers` on purpose, as we should not log user-data.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java:
##########
@@ -330,7 +330,8 @@ private void reprocessState(final List<TopicPartition>
topicPartitions,
Thread.currentThread().getName(),
globalProcessorContext.taskId().toString(),
globalProcessorContext.metrics()
- )
+ ),
+ null
Review Comment:
Side cleanup as discussed on some other PR.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java:
##########
@@ -202,7 +202,7 @@ public void process(final Record<KIn, VIn> record) {
} catch (final FailedProcessingException | TaskCorruptedException |
TaskMigratedException e) {
// Rethrow exceptions that should not be handled here
throw e;
- } catch (final RuntimeException e) {
+ } catch (final RuntimeException processingException) {
Review Comment:
side cleanup: use proper variable names
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java:
##########
@@ -205,7 +207,7 @@ public <K, V> void send(final String topic,
key,
keySerializer,
exception);
- } catch (final Exception exception) {
+ } catch (final RuntimeException serializationException) {
Review Comment:
side cleanup: use proper variable names
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -815,19 +815,25 @@ record = null;
return true;
}
- private void handleException(final Throwable e) {
- final StreamsException error = new StreamsException(
+ private void handleException(final Throwable originalException) {
+ handleException(
String.format(
- "Exception caught in process. taskId=%s, processor=%s,
topic=%s, partition=%d, offset=%d, stacktrace=%s",
Review Comment:
given what we pass in `originalException` into `StreamsException, it seems
not necessary to include the stracktrace in the error message.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -915,6 +917,13 @@ public void punctuate(final ProcessorNode<?, ?, ?, ?> node,
try {
maybeMeasureLatency(() -> punctuator.punctuate(timestamp), time,
punctuateLatencySensor);
+ } catch (final TimeoutException timeoutException) {
Review Comment:
Totally unrelated (might actually to good to put into different PR, but want
to avoid that I forget it).
Seems we forgot to handle this case entirely... we might call
`context.forward()` inside of `punctuate()` and thus need to handle the same
error cases as for `node.process()`...
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java:
##########
@@ -293,27 +295,33 @@ private <K, V> void handleException(final
ProductionExceptionHandler.Serializati
final Long timestamp,
final String processorNodeId,
final InternalProcessorContext<Void,
Void> context,
- final Exception exception) {
+ final RuntimeException
serializationException) {
+ log.debug(String.format("Error serializing record for topic %s",
topic), serializationException);
+
+ final DefaultErrorHandlerContext errorHandlerContext = new
DefaultErrorHandlerContext(
+ null, // only required to pass for DeserializationExceptionHandler
+ context.recordContext().topic(),
+ context.recordContext().partition(),
+ context.recordContext().offset(),
+ context.recordContext().headers(),
+ processorNodeId,
+ taskId
+ );
final ProducerRecord<K, V> record = new ProducerRecord<>(topic,
partition, timestamp, key, value, headers);
- final ProductionExceptionHandlerResponse response;
-
- log.debug(String.format("Error serializing record to topic %s",
topic), exception);
+ final ProductionExceptionHandlerResponse response;
try {
- final DefaultErrorHandlerContext errorHandlerContext = new
DefaultErrorHandlerContext(
- null, // only required to pass for
DeserializationExceptionHandler
- context.recordContext().topic(),
- context.recordContext().partition(),
- context.recordContext().offset(),
- context.recordContext().headers(),
- processorNodeId,
- taskId
+ response =
productionExceptionHandler.handleSerializationException(errorHandlerContext,
record, serializationException, origin);
+ } catch (final RuntimeException fatalUserException) {
+ log.error(
+ String.format(
+ "Production error callback failed after serialization
error for record %s: %s",
+ origin.toString().toLowerCase(Locale.ROOT),
+ errorHandlerContext
+ ),
+ serializationException
);
- 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;
+ throw new FailedProcessingException("Fatal user code error in
production error callback", fatalUserException);
Review Comment:
fix: inside `send()` is seem incorrect to call `recordSendError()` which we
only use in from the `producer.send(..., Calback)` which is called async. Here
inside `RecordCollector.send()` (before we hit `producer.send()`) can should
rather throw directly
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordDeserializer.java:
##########
@@ -69,47 +70,42 @@ ConsumerRecord<Object, Object> deserialize(final
ProcessorContext<?, ?> processo
rawRecord.headers(),
Optional.empty()
);
- } catch (final Exception deserializationException) {
+ } catch (final RuntimeException deserializationException) {
handleDeserializationFailure(deserializationExceptionHandler,
processorContext, deserializationException, rawRecord, log,
droppedRecordsSensor, sourceNode().name());
return null; // 'handleDeserializationFailure' would either throw
or swallow -- if we swallow we need to skip the record by returning 'null'
}
}
public static void handleDeserializationFailure(final
DeserializationExceptionHandler deserializationExceptionHandler,
final ProcessorContext<?,
?> processorContext,
- final Exception
deserializationException,
- final
ConsumerRecord<byte[], byte[]> rawRecord,
- final Logger log,
- final Sensor
droppedRecordsSensor) {
- handleDeserializationFailure(deserializationExceptionHandler,
processorContext, deserializationException, rawRecord, log,
droppedRecordsSensor, null);
Review Comment:
Side cleanup as discussed on some other PR
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -815,19 +815,25 @@ record = null;
return true;
}
- private void handleException(final Throwable e) {
- final StreamsException error = new StreamsException(
+ private void handleException(final Throwable originalException) {
+ handleException(
String.format(
- "Exception caught in process. taskId=%s, processor=%s,
topic=%s, partition=%d, offset=%d, stacktrace=%s",
+ "Exception caught in process. taskId=%s, processor=%s,
topic=%s, partition=%d, offset=%d",
id(),
processorContext.currentNode().name(),
record.topic(),
record.partition(),
- record.offset(),
- getStacktraceString(e)
+ record.offset()
),
- e
- );
+ originalException);
+ }
+
+ private void handleException(final String errorMessage, final Throwable
originalException) {
Review Comment:
this is an attempt to preserve the error message we create when a handler
throws by itself.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java:
##########
@@ -406,17 +414,31 @@ private void recordSendError(final String topic,
taskId
);
- if (productionExceptionHandler.handle(errorHandlerContext,
serializedRecord, exception) == ProductionExceptionHandlerResponse.FAIL) {
+ final ProductionExceptionHandlerResponse response;
+ try {
+ response =
productionExceptionHandler.handle(errorHandlerContext, serializedRecord,
productionException);
+ } catch (final RuntimeException fatalUserException) {
Review Comment:
new: handle exception from `ProductionExceptionHandler` callback
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java:
##########
@@ -372,24 +380,24 @@ private <V> StreamsException
createStreamsExceptionForValueClassCastException(fi
}
private void recordSendError(final String topic,
- final Exception exception,
+ final Exception productionException,
Review Comment:
side cleanup: use proper variable names
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
