philipnee commented on code in PR #13477:
URL: https://github.com/apache/kafka/pull/13477#discussion_r1167093584
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java:
##########
@@ -1391,21 +1390,57 @@ public void shouldNotCloseInternalProducerForNonEOS() {
}
@Test
- public void
testCollectorFlush_ThrowsStreamsExceptionUsingDefaultExceptionHandler() {
- final ErrorStringSerializer errorSerializer = new
ErrorStringSerializer();
- final RecordCollector collector = newRecordCollector(new
DefaultProductionExceptionHandler());
- collector.initialize();
- collector.send(topic, "key", "val", null, 0, null, stringSerializer,
errorSerializer, sinkNodeName, context);
- assertThrows(StreamsException.class, collector::flush);
+ public void shouldThrowStreamsExceptionUsingDefaultExceptionHandler() {
+ try (final ErrorStringSerializer errorSerializer = new
ErrorStringSerializer()) {
+ final RecordCollector collector = newRecordCollector(new
DefaultProductionExceptionHandler());
+ collector.initialize();
+
+ final StreamsException error = assertThrows(
+ StreamsException.class,
+ () -> collector.send(topic, "key", "val", null, 0, null,
stringSerializer, errorSerializer, sinkNodeName, context)
+ );
+
+ assertThat(error.getCause(),
instanceOf(SerializationException.class));
+ }
}
@Test
- public void
testCollectorFlush_DoesNotThrowStreamsExceptionUsingAlwaysContinueExceptionHandler()
{
- final ErrorStringSerializer errorSerializer = new
ErrorStringSerializer();
- final RecordCollector collector = newRecordCollector(new
AlwaysContinueProductionExceptionHandler());
- collector.initialize();
- collector.send(topic, "key", "val", null, 0, null, errorSerializer,
stringSerializer, sinkNodeName, context);
- assertDoesNotThrow(collector::flush);
+ public void shouldDropRecordExceptionUsingAlwaysContinueExceptionHandler()
{
+ try (final ErrorStringSerializer errorSerializer = new
ErrorStringSerializer()) {
+ final RecordCollector collector = newRecordCollector(new
AlwaysContinueProductionExceptionHandler());
+ collector.initialize();
+
+ collector.send(topic, "key", "val", null, 0, null,
errorSerializer, stringSerializer, sinkNodeName, context);
+
+ assertThat(mockProducer.history().isEmpty(), equalTo(true));
+ assertThat(
+ streamsMetrics.metrics().get(new MetricName(
+ "dropped-records-total",
+ "stream-task-metrics",
+ "The total number of dropped records",
+ mkMap(
+ mkEntry("thread-id", Thread.currentThread().getName()),
+ mkEntry("task-id", taskId.toString())
+ ))).metricValue(),
+ equalTo(1.0)
+ );
+ }
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ @Test
+ public void shouldNotCallProductionExceptionHandlerOnClassCastException() {
+ try (final ErrorStringSerializer errorSerializer = new
ErrorStringSerializer()) {
+ final RecordCollector collector = newRecordCollector(new
AlwaysContinueProductionExceptionHandler());
+ collector.initialize();
+
Review Comment:
Here's a minor PR: https://github.com/apache/kafka/pull/13576
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
