You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mj...@apache.org on 2019/10/17 07:24:19 UTC
[kafka] branch 2.4 updated: KAFKA-8884: class cast exception
improvement (#7309)
This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch 2.4
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/2.4 by this push:
new de5254d KAFKA-8884: class cast exception improvement (#7309)
de5254d is described below
commit de5254db7284bbcb44661518c34ff2a38f8256c1
Author: Antony Stubbs <an...@gmail.com>
AuthorDate: Thu Oct 17 08:00:39 2019 +0100
KAFKA-8884: class cast exception improvement (#7309)
Reviewers: John Roesler <jo...@confluent.io>, Matthias J. Sax <ma...@confluent.io>
---
.../streams/processor/internals/ProcessorNode.java | 18 +++++++-
.../streams/processor/internals/SinkNode.java | 4 +-
.../processor/internals/ProcessorNodeTest.java | 52 +++++++++++++++++++++-
3 files changed, 70 insertions(+), 4 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
index 6e10e83..3a5a7db 100644
--- 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
@@ -114,7 +114,23 @@ public class ProcessorNode<K, V> {
public void process(final K key, final V value) {
final long startNs = time.nanoseconds();
- processor.process(key, value);
+ try {
+ processor.process(key, value);
+ } catch (final ClassCastException e) {
+ final String keyClass = key == null ? "unknown because key is null" : key.getClass().getName();
+ final String valueClass = value == null ? "unknown because value is null" : value.getClass().getName();
+ throw new StreamsException(String.format("ClassCastException invoking Processor. Do the Processor's "
+ + "input types match the deserialized types? Check the Serde setup and change the default Serdes in "
+ + "StreamConfig or provide correct Serdes via method parameters. Make sure the Processor can accept "
+ + "the deserialized input of type key: %s, and value: %s.%n"
+ + "Note that although incorrect Serdes are a common cause of error, the cast exception might have "
+ + "another cause (in user code, for example). For example, if a processor wires in a store, but casts "
+ + "the generics incorrectly, a class cast exception could be raised during processing, but the "
+ + "cause would not be wrong Serdes.",
+ keyClass,
+ valueClass),
+ e);
+ }
nodeMetrics.nodeProcessTimeSensor.record(time.nanoseconds() - startNs);
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java
index 73bffc8..b94291b 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java
@@ -91,9 +91,9 @@ public class SinkNode<K, V> extends ProcessorNode<K, V> {
final String keyClass = key == null ? "unknown because key is null" : key.getClass().getName();
final String valueClass = value == null ? "unknown because value is null" : value.getClass().getName();
throw new StreamsException(
- String.format("A serializer (key: %s / value: %s) is not compatible to the actual key or value type " +
+ String.format("ClassCastException while producing data to a sink topic. A serializer (key: %s / value: %s) is not compatible to the actual key or value type " +
"(key type: %s / value type: %s). Change the default Serdes in StreamConfig or " +
- "provide correct Serdes via method parameters.",
+ "provide correct Serdes via method parameters (for example if using the DSL, `#to(String topic, Produced<K, V> produced)` with `Produced.keySerde(WindowedSerdes.timeWindowedSerdeFrom(String.class))`).",
keySerializer.getClass().getName(),
valSerializer.getClass().getName(),
keyClass,
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java
index 42ad94e..3ae3852 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java
@@ -16,10 +16,18 @@
*/
package org.apache.kafka.streams.processor.internals;
+import java.util.Arrays;
+import java.util.Properties;
import org.apache.kafka.common.metrics.JmxReporter;
import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.TestInputTopic;
+import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.TopologyTestDriver;
import org.apache.kafka.streams.errors.DefaultProductionExceptionHandler;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.processor.Processor;
@@ -33,7 +41,11 @@ import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
+import static org.hamcrest.CoreMatchers.containsString;
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
public class ProcessorNodeTest {
@@ -52,7 +64,7 @@ public class ProcessorNodeTest {
node.close();
}
- private static class ExceptionalProcessor implements Processor {
+ private static class ExceptionalProcessor implements Processor<Object, Object> {
@Override
public void init(final ProcessorContext context) {
throw new RuntimeException();
@@ -142,4 +154,42 @@ public class ProcessorNodeTest {
groupName, threadId, context.taskId().toString(), node.name())));
}
+ @Test
+ public void testTopologyLevelClassCastException() {
+ final Properties props = new Properties();
+ props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test");
+ props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:1234");
+ // Serdes configuration is missing (default will be used which don't match the DSL below), which will trigger the new exception
+ final StreamsBuilder builder = new StreamsBuilder();
+
+ builder.<String, String>stream("streams-plaintext-input")
+ .flatMapValues(value -> {
+ return Arrays.asList("");
+ });
+ final Topology topology = builder.build();
+
+ final TopologyTestDriver testDriver = new TopologyTestDriver(topology, props);
+ final TestInputTopic<String, String> topic = testDriver.createInputTopic("streams-plaintext-input", new StringSerializer(), new StringSerializer());
+
+ final StreamsException se = assertThrows(StreamsException.class, () -> topic.pipeInput("a-key", "a value"));
+ final String msg = se.getMessage();
+ assertTrue("Error about class cast with serdes", msg.contains("ClassCastException"));
+ assertTrue("Error about class cast with serdes", msg.contains("Serdes"));
+ }
+
+ private static class ClassCastProcessor extends ExceptionalProcessor {
+ @Override
+ public void process(final Object key, final Object value) {
+ throw new ClassCastException("Incompatible types simulation exception.");
+ }
+ }
+
+ @Test
+ public void testTopologyLevelClassCastExceptionDirect() {
+ final ProcessorNode<Object, Object> node = new ProcessorNode<Object, Object>("name", new ClassCastProcessor(), Collections.<String>emptySet());
+ final StreamsException se = assertThrows(StreamsException.class, () -> node.process("aKey", "aValue"));
+ assertThat(se.getCause(), instanceOf(ClassCastException.class));
+ assertThat(se.getMessage(), containsString("default Serdes"));
+ assertThat(se.getMessage(), containsString("input types"));
+ }
}