You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "sebastienviale (via GitHub)" <gi...@apache.org> on 2024/06/24 08:06:00 UTC
[PR] Kafka 16448 production add error handler context [kafka]
sebastienviale opened a new pull request, #16433:
URL: https://github.com/apache/kafka/pull/16433
This PR is part of [KAFKA-16448](https://issues.apache.org/jira/browse/KAFKA-16448) 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
Jira: https://issues.apache.org/jira/browse/KAFKA-16448.
Contributors
@Dabz
@sebastienviale
@loicgreffier
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #16433:
URL: https://github.com/apache/kafka/pull/16433#discussion_r1693449190
##########
streams/src/main/java/org/apache/kafka/streams/errors/ProductionExceptionHandler.java:
##########
@@ -30,22 +30,60 @@ public interface ProductionExceptionHandler extends Configurable {
*
* @param record The record that failed to produce
* @param exception The exception that occurred during production
+ * @deprecated 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
* fail the process.
*
* @param record the record that failed to serialize
* @param exception the exception that occurred during serialization
+ * @deprecated Use {@link #handle(ErrorHandlerContext, ProducerRecord, Exception)} instead
Review Comment:
```suggestion
* @deprecated Since 3.9. Use {@link #handle(ErrorHandlerContext, ProducerRecord, Exception)} instead.
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #16433:
URL: https://github.com/apache/kafka/pull/16433#discussion_r1694824990
##########
streams/src/main/java/org/apache/kafka/streams/errors/ProductionExceptionHandler.java:
##########
@@ -30,22 +30,60 @@ public interface ProductionExceptionHandler extends Configurable {
*
* @param record The record that failed to produce
* @param exception The exception that occurred during production
+ * @deprecated 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();
Review Comment:
KIP has been updated
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "loicgreffier (via GitHub)" <gi...@apache.org>.
loicgreffier commented on code in PR #16433:
URL: https://github.com/apache/kafka/pull/16433#discussion_r1689395502
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java:
##########
@@ -217,16 +218,32 @@ public <K, V> void send(final String topic,
valueClass),
exception);
} catch (final Exception exception) {
+ final ProductionExceptionHandler.SerializationExceptionOrigin origin;
+ if (keyBytes == null) {
+ origin = ProductionExceptionHandler.SerializationExceptionOrigin.KEY;
+ } else {
+ origin = ProductionExceptionHandler.SerializationExceptionOrigin.VALUE;
+ }
Review Comment:
The update has been pushed!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #16433:
URL: https://github.com/apache/kafka/pull/16433
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #16433:
URL: https://github.com/apache/kafka/pull/16433#discussion_r1689627424
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java:
##########
@@ -1547,11 +1565,32 @@ public void shouldDropRecordExceptionUsingAlwaysContinueExceptionHandler() {
}
}
+ @Test
+ public void shouldThrowStreamsExceptionWhenValueSerializationFailedAndProductionExceptionHandlerRepliesWithFail() {
Review Comment:
done, unit test added
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "loicgreffier (via GitHub)" <gi...@apache.org>.
loicgreffier commented on code in PR #16433:
URL: https://github.com/apache/kafka/pull/16433#discussion_r1688194251
##########
streams/src/main/java/org/apache/kafka/streams/errors/ProductionExceptionHandler.java:
##########
@@ -30,22 +30,60 @@ public interface ProductionExceptionHandler extends Configurable {
*
* @param record The record that failed to produce
* @param exception The exception that occurred during production
+ * @deprecated 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
* fail the process.
*
- * @param record the record that failed to serialize
- * @param exception the exception that occurred during serialization
+ * @param record the record that failed to serialize
+ * @param exception the exception that occurred during serialization
Review Comment:
Done
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #16433:
URL: https://github.com/apache/kafka/pull/16433#discussion_r1693448882
##########
streams/src/main/java/org/apache/kafka/streams/errors/ProductionExceptionHandler.java:
##########
@@ -30,22 +30,60 @@ public interface ProductionExceptionHandler extends Configurable {
*
* @param record The record that failed to produce
* @param exception The exception that occurred during production
+ * @deprecated Use {@link #handle(ErrorHandlerContext, ProducerRecord, Exception)} instead
Review Comment:
```suggestion
* @deprecated Since 3.9. Use {@link #handle(ErrorHandlerContext, ProducerRecord, Exception)} instead.
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #16433:
URL: https://github.com/apache/kafka/pull/16433#discussion_r1693451743
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java:
##########
@@ -285,15 +280,95 @@ public <K, V> void send(final String topic,
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);
}
});
}
- 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(),
+ context.recordContext().rawRecord().key(),
+ context.recordContext().rawRecord().value(),
+ 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, V> StreamsException createStreamsExceptionForClassCastException(final String topic,
Review Comment:
Given that we introduce a helper and have two dedicated try-catch blocks for key and value now, it seems we can tailor this method to key or value similar to what we do for `handleException` instead of this generic one covering both at once?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #16433:
URL: https://github.com/apache/kafka/pull/16433#issuecomment-2257434274
Thanks for the PR.
Merged to `trunk` and cherry-picked to `3.9` branch.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #16433:
URL: https://github.com/apache/kafka/pull/16433#discussion_r1694821522
##########
streams/src/main/java/org/apache/kafka/streams/errors/ProductionExceptionHandler.java:
##########
@@ -30,22 +30,60 @@ public interface ProductionExceptionHandler extends Configurable {
*
* @param record The record that failed to produce
* @param exception The exception that occurred during production
+ * @deprecated Use {@link #handle(ErrorHandlerContext, ProducerRecord, Exception)} instead
Review Comment:
done
##########
streams/src/main/java/org/apache/kafka/streams/errors/ProductionExceptionHandler.java:
##########
@@ -30,22 +30,60 @@ public interface ProductionExceptionHandler extends Configurable {
*
* @param record The record that failed to produce
* @param exception The exception that occurred during production
+ * @deprecated 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
* fail the process.
*
* @param record the record that failed to serialize
* @param exception the exception that occurred during serialization
+ * @deprecated Use {@link #handle(ErrorHandlerContext, ProducerRecord, Exception)} instead
Review Comment:
done
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #16433:
URL: https://github.com/apache/kafka/pull/16433#discussion_r1693454083
##########
streams/src/test/java/org/apache/kafka/streams/errors/AlwaysContinueProductionExceptionHandler.java:
##########
@@ -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 {
Review Comment:
But `AlwaysContinueProductionExceptionHandler` is a public built-in handler users can configure... -- it's public API, and thus has nothing to do with `ProductionExceptionHandlerMock` and our internal testing.
It's a build-in alternative to the default handler.
If we really want to no have it any longer, we would need to deprecate -- but we cannot just remove public API. But I don't see why we would want to deprecate and remove it?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #16433:
URL: https://github.com/apache/kafka/pull/16433#discussion_r1692270936
##########
streams/src/test/java/org/apache/kafka/streams/errors/AlwaysContinueProductionExceptionHandler.java:
##########
@@ -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 {
Review Comment:
Why does this PR remove this class?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #16433:
URL: https://github.com/apache/kafka/pull/16433#discussion_r1696213764
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java:
##########
@@ -285,15 +280,95 @@ public <K, V> void send(final String topic,
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);
}
});
}
- 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(),
+ context.recordContext().rawRecord().key(),
+ context.recordContext().rawRecord().value(),
+ 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, V> StreamsException createStreamsExceptionForClassCastException(final String topic,
Review Comment:
Yes, and I merged this PR to make progress. However, I think, we can unify both methods into a single one, passing in `"key"` or `"value"` as parameter to avoid code duplication (similar to the "origin" parameter use in `handleException(...)`)
The generic type would could be updated to `<KV>` for "key or value type" or just to `<T>` if you prefer.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java:
##########
@@ -285,15 +276,104 @@ public <K, V> void send(final String topic,
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);
}
});
}
- 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,
Review Comment:
Nit: missing newline
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "loicgreffier (via GitHub)" <gi...@apache.org>.
loicgreffier commented on code in PR #16433:
URL: https://github.com/apache/kafka/pull/16433#discussion_r1688203643
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java:
##########
@@ -1705,4 +1742,70 @@ public byte[] serialize(final String topic, final Headers headers, final String
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(expectedContext.recordContext().rawRecord().key(), context.sourceRawKey());
+ assertEquals(expectedContext.recordContext().rawRecord().value(), context.sourceRawValue());
Review Comment:
Done
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "loicgreffier (via GitHub)" <gi...@apache.org>.
loicgreffier commented on code in PR #16433:
URL: https://github.com/apache/kafka/pull/16433#discussion_r1688257379
##########
streams/src/main/java/org/apache/kafka/streams/errors/DefaultProductionExceptionHandler.java:
##########
@@ -26,7 +26,8 @@
*/
public class DefaultProductionExceptionHandler implements ProductionExceptionHandler {
@Override
- public ProductionExceptionHandlerResponse handle(final ProducerRecord<byte[], byte[]> record,
+ public ProductionExceptionHandlerResponse handle(final ErrorHandlerContext context,
Review Comment:
@cadonna I updated KIP-1033 with a word regarding deserialization and production handler implementations
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "loicgreffier (via GitHub)" <gi...@apache.org>.
loicgreffier commented on code in PR #16433:
URL: https://github.com/apache/kafka/pull/16433#discussion_r1692667410
##########
streams/src/test/java/org/apache/kafka/streams/errors/AlwaysContinueProductionExceptionHandler.java:
##########
@@ -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 {
Review Comment:
@mjsax
We have created a parameterized `ProductionExceptionHandler` (`ProductionExceptionHandlerMock`) in the `RecordCollectorTest` class that additionally does some assertions, including the origin of the exception.
Thus `AlwaysContinueProductionExceptionHandler` was redundant with `ProductionExceptionHandlerMock`, it has been replaced in this PR while updating and adding some additional unit tests
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "loicgreffier (via GitHub)" <gi...@apache.org>.
loicgreffier commented on code in PR #16433:
URL: https://github.com/apache/kafka/pull/16433#discussion_r1688202709
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java:
##########
@@ -217,16 +218,32 @@ public <K, V> void send(final String topic,
valueClass),
exception);
} catch (final Exception exception) {
+ final ProductionExceptionHandler.SerializationExceptionOrigin origin;
+ if (keyBytes == null) {
+ origin = ProductionExceptionHandler.SerializationExceptionOrigin.KEY;
+ } else {
+ origin = ProductionExceptionHandler.SerializationExceptionOrigin.VALUE;
+ }
Review Comment:
Done.
I had to remove the `final` from `keyBytes` and `valueBytes` and initialize them when moving the `ClassCastException` code to a dedicated method
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "loicgreffier (via GitHub)" <gi...@apache.org>.
loicgreffier commented on code in PR #16433:
URL: https://github.com/apache/kafka/pull/16433#discussion_r1688206735
##########
streams/src/main/java/org/apache/kafka/streams/errors/DefaultProductionExceptionHandler.java:
##########
@@ -26,7 +26,8 @@
*/
public class DefaultProductionExceptionHandler implements ProductionExceptionHandler {
@Override
- public ProductionExceptionHandlerResponse handle(final ProducerRecord<byte[], byte[]> record,
+ public ProductionExceptionHandlerResponse handle(final ErrorHandlerContext context,
Review Comment:
True, we did it for the deserialization exception handler implementations. I'm wondering why we did not do the same for the production.. 😄
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #16433:
URL: https://github.com/apache/kafka/pull/16433#discussion_r1689346485
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java:
##########
@@ -217,16 +218,32 @@ public <K, V> void send(final String topic,
valueClass),
exception);
} catch (final Exception exception) {
+ final ProductionExceptionHandler.SerializationExceptionOrigin origin;
+ if (keyBytes == null) {
+ origin = ProductionExceptionHandler.SerializationExceptionOrigin.KEY;
+ } else {
+ origin = ProductionExceptionHandler.SerializationExceptionOrigin.VALUE;
+ }
Review Comment:
If you want to keep the `final` and make the code a bit better readable, you could rename `handleClassCastException()` to `createStreamsExceptionForClassCastException()` and return the `StreamsException` from there. Then you can do
```java
} catch (final ClassCastException exception) {
throw createStreamsExceptionForClassCastException()(
topic,
key,
value,
keySerializer,
valueSerializer,
exception);
```
With this it is clear that the `catch`-branch terminates the method.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #16433:
URL: https://github.com/apache/kafka/pull/16433#discussion_r1694822424
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java:
##########
@@ -285,15 +280,95 @@ public <K, V> void send(final String topic,
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);
}
});
}
- 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(),
+ context.recordContext().rawRecord().key(),
+ context.recordContext().rawRecord().value(),
+ 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, V> StreamsException createStreamsExceptionForClassCastException(final String topic,
Review Comment:
done, I hope this what you are waiting !
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #16433:
URL: https://github.com/apache/kafka/pull/16433#discussion_r1689396883
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java:
##########
@@ -1547,11 +1565,32 @@ public void shouldDropRecordExceptionUsingAlwaysContinueExceptionHandler() {
}
}
+ @Test
+ public void shouldThrowStreamsExceptionWhenValueSerializationFailedAndProductionExceptionHandlerRepliesWithFail() {
Review Comment:
I am missing the version of this test where the key serialization fails.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "loicgreffier (via GitHub)" <gi...@apache.org>.
loicgreffier commented on code in PR #16433:
URL: https://github.com/apache/kafka/pull/16433#discussion_r1697519314
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java:
##########
@@ -285,15 +280,95 @@ public <K, V> void send(final String topic,
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);
}
});
}
- 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(),
+ context.recordContext().rawRecord().key(),
+ context.recordContext().rawRecord().value(),
+ 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, V> StreamsException createStreamsExceptionForClassCastException(final String topic,
Review Comment:
@mjsax Addressing it in https://github.com/apache/kafka/pull/16736
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java:
##########
@@ -285,15 +276,104 @@ public <K, V> void send(final String topic,
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);
}
});
}
- 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,
Review Comment:
@mjsax Addressing it in https://github.com/apache/kafka/pull/16736
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #16433:
URL: https://github.com/apache/kafka/pull/16433#discussion_r1687751846
##########
streams/src/main/java/org/apache/kafka/streams/errors/DefaultProductionExceptionHandler.java:
##########
@@ -26,7 +26,8 @@
*/
public class DefaultProductionExceptionHandler implements ProductionExceptionHandler {
@Override
- public ProductionExceptionHandlerResponse handle(final ProducerRecord<byte[], byte[]> record,
+ public ProductionExceptionHandlerResponse handle(final ErrorHandlerContext context,
Review Comment:
I do not think we can modify this without deprecating the old method and mentioning it in the KIP since it is part of the public API.
I am aware that this method is only called internally.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java:
##########
@@ -1705,4 +1742,70 @@ public byte[] serialize(final String topic, final Headers headers, final String
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(expectedContext.recordContext().rawRecord().key(), context.sourceRawKey());
+ assertEquals(expectedContext.recordContext().rawRecord().value(), context.sourceRawValue());
Review Comment:
I think you need to use `assertArrayEquals()` here to verify the byte arrays.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java:
##########
@@ -217,16 +218,32 @@ public <K, V> void send(final String topic,
valueClass),
exception);
} catch (final Exception exception) {
+ final ProductionExceptionHandler.SerializationExceptionOrigin origin;
+ if (keyBytes == null) {
+ origin = ProductionExceptionHandler.SerializationExceptionOrigin.KEY;
+ } else {
+ origin = ProductionExceptionHandler.SerializationExceptionOrigin.VALUE;
+ }
Review Comment:
I think, it would be better to have something like:
```java
try {
keyBytes = keySerializer.serialize(topic, headers, key);
} catch (final ClassCastException exception) {
handleClassCastException(...)
} catch (final Exception exception) {
handleException(ProductionExceptionHandler.SerializationExceptionOrigin.KEY, ...)
}
try {
valBytes = valueSerializer.serialize(topic, headers, value);
} catch (final ClassCastException exception) {
handleClassCastException(...)
} catch (final Exception exception) {
handleException(ProductionExceptionHandler.SerializationExceptionOrigin.VALUE, ...)
}
```
##########
streams/src/main/java/org/apache/kafka/streams/errors/ProductionExceptionHandler.java:
##########
@@ -30,22 +30,60 @@ public interface ProductionExceptionHandler extends Configurable {
*
* @param record The record that failed to produce
* @param exception The exception that occurred during production
+ * @deprecated 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
* fail the process.
*
- * @param record the record that failed to serialize
- * @param exception the exception that occurred during serialization
+ * @param record the record that failed to serialize
+ * @param exception the exception that occurred during serialization
Review Comment:
Could you please revert the formatting changes in this two lines?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #16433:
URL: https://github.com/apache/kafka/pull/16433#discussion_r1696207479
##########
streams/src/test/java/org/apache/kafka/streams/errors/AlwaysContinueProductionExceptionHandler.java:
##########
@@ -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 {
Review Comment:
Ah. You are right. My bad. I guess I mixed it up with the two built-in handler for `DeserializationExceptionHandler`, but we only have the "alway fail" build-in handler for the production case...
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #16433:
URL: https://github.com/apache/kafka/pull/16433#discussion_r1693449812
##########
streams/src/main/java/org/apache/kafka/streams/errors/ProductionExceptionHandler.java:
##########
@@ -30,22 +30,60 @@ public interface ProductionExceptionHandler extends Configurable {
*
* @param record The record that failed to produce
* @param exception The exception that occurred during production
+ * @deprecated 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();
Review Comment:
KIP says `NotImplementedException` -- we should update the KIP -- same for the other method below.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Add ErrorHandlerContext in production exception handler [kafka]
Posted by "loicgreffier (via GitHub)" <gi...@apache.org>.
loicgreffier commented on code in PR #16433:
URL: https://github.com/apache/kafka/pull/16433#discussion_r1694707610
##########
streams/src/test/java/org/apache/kafka/streams/errors/AlwaysContinueProductionExceptionHandler.java:
##########
@@ -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 {
Review Comment:
@mjsax Unless I'm wrong, `AlwaysContinueProductionExceptionHandler` belongs to `src/test/java` and is only used for testing purpose in the `RecordCollectorTest`.
Given that it is only use for testing, should it be consider public? If yes, we can obviously restore the class
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org