Hi, I’m using Kafka Streams (0.10.0) with JRuby, most of my scripts/nodes are
working well at this point, except for one which is using reduceByKey.
This is the first time I’m trying to use the local state store so it’s possible
there’s something misconfigured, I’m not sure. My config is pretty vanilla and
minimal.
My debugging so far shows that reduceByKey is passing nil/null values to my
Deserializer. I wasn’t expecting this and my Deserializer is currently raising
exceptions in this case.
I guess I’d like to know — is this normal, expected behavior? If so, why, and
what does it mean, and how am I meant to handle it?
If not, any idea why it might be happening?
My stack trace is pretty crazy due to JRuby (there’s probably a way to filter
it better but I’m new to JRuby, sorry), but here are the most salient lines:
at RUBY.deserialize(uri:classloader:/lib/avro_utils/avro_kafka_serde.rb:31)
at
org.jruby.internal.runtime.methods.MixedModeIRMethod.call(org/jruby/internal/runtime/methods/MixedModeIRMethod.java:126)
at
KafkaAvroHashSerdes$$Deserializer_1394592798.deserialize(KafkaAvroHashSerdes$$Deserializer_1394592798.gen:13)
at
org.apache.kafka.streams.state.StateSerdes.valueFrom(org/apache/kafka/streams/state/StateSerdes.java:156)
at
org.apache.kafka.streams.state.internals.RocksDBStore.get(org/apache/kafka/streams/state/internals/RocksDBStore.java:241)
at
org.apache.kafka.streams.state.internals.MeteredKeyValueStore.get(org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java:100)
at
org.apache.kafka.streams.kstream.internals.KStreamReduce$KStreamReduceProcessor.process(org/apache/kafka/streams/kstream/internals/KStreamReduce.java:70)
You can see the entire thing here:
https://gist.github.com/aviflax/3428cdbaa18aca9bf0a958c6d5eac2bd
Thank you!
Avi
