Hi,
I have ran into a strange issue when using the kafka producer.
I got the following exception:
Caused by: java.lang.IllegalArgumentException: Invalid partition given
with record: 5 is not in the range [0...2].
at
org.apache.kafka.clients.producer.internals.Partitioner.partition(Partitioner.java:52)
at
org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:333)
at
org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducerBase.invoke(FlinkKafkaProducerBase.java:260)
at
org.apache.flink.streaming.api.operators.StreamSink.processElement(StreamSink.java:39)
at
org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:351)
... 19 more
It looks like the partition number returned by the partitioner is out of
range, which is pretty much impossible as I am doing a very straightforward
mod-hash partitioning logic:
return (int) Math.abs(element.hashCode() % numPartitions);
This topic is supposed to have 16 partitions and it shows it only has 3 and
thus the out of bound error for the partitioner. I am not completely sure
what happened with the Kafka cluster that could have caused this issue to
appear.
This issue also appears if I restore from a savepoint/checkpoint before
this happened, it runs for a couple of minutes then crashes. (I am also not
completely sure why it doesnt crash immediately)
Has anyone seen something similar before?
Thanks!
Gyula
