Whoops, I said I'd put the specific exception at the bottom of the
e-mail. It probably isn't the important part of this thread, but
might suggest when this situation can occur. Also of note, this is
occurring on Kafka Streams 0.10.2.1.
20:56:07.061 [StreamThread-3] ERROR o.a.k.s.p.internals.StreamThread -
stream-thread [StreamThread-3] Failed to commit StreamTask 0_4 state:
org.apache.kafka.streams.errors.StreamsException: task [0_4] exception
caught when producing
at
org.apache.kafka.streams.processor.internals.RecordCollectorImpl.checkForException(RecordCollectorImpl.java:121)
at
org.apache.kafka.streams.processor.internals.RecordCollectorImpl.flush(RecordCollectorImpl.java:129)
at
org.apache.kafka.streams.processor.internals.StreamTask$1.run(StreamTask.java:76)
at
org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:188)
at
org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:280)
at
org.apache.kafka.streams.processor.internals.StreamThread.commitOne(StreamThread.java:807)
at
org.apache.kafka.streams.processor.internals.StreamThread.commitAll(StreamThread.java:794)
at
org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:769)
at
org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:647)
at
org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:361)
Caused by: org.apache.kafka.common.errors.TimeoutException: Failed to
update metadata after 60000 ms.
On Fri, May 19, 2017 at 9:47 AM, Mathieu Fenniak
<mathieu.fenn...@replicon.com> wrote:
> Hi Kafka devs,
>
> This morning I observed a specific Kafka Streams aggregation that
> ended up with an incorrect computed output after a Kafka Streams
> thread crashed with an unhandled exception.
>
> The topology is pretty simple -- a single KTable source, group by a
> field in the value, aggregate that adds up another field, output to a
> topic.
>
> Here's the sequence of events that appears to have occurred:
>
> 1. A new record (record A) is received by the source KTable, and put
> in the KTable RocksDB state store.
>
> 2. While processing record A, an exception happens preventing
> producing to Kafka. (specific exception at end of e-mail).
>
> 3. The stream thread throws an unhandled exception and stops.
>
> 4. The state stores are closed and flushed. Record A is now in the
> local state store.
>
> 5. The consumer group rebalances.
>
> 6. A different thread, in the same process, on the same host, picks up the
> task.
>
> 7. New thread initializes its state store for the KTable, but it's on
> the same host as the original thread, so it still contains the k/v for
> record A.
>
> 8. New thread resumes consuming at the last committed offset, which is
> before record A.
>
> 9. When processing record A, the new thread reads the value that was
> written to the state store in step #1 by record A's key.
>
> 10. The repartition map receives a Change with both an oldValue and a
> newValue, and forwards a Change(null, v) and Change(v, null)
>
> 11. The aggregation ends up both subtracting and adding the value of
> record A, resulting in an incorrect output.
>
> As a result of this sequence, my aggregate output went from a value of
> 0, to negative (subtracting record A), to 0. And stayed there.
>
> Does this seem like a feasible series of events? Is this a bug in KS,
> or, is it behavior that maybe can't be improved without exactly-once?
> I'd think the best behavior would be for writes to the RocksDB state
> store to be transactional and only commit when the producer commits,
> but, there's a lot of overhead involved in that.
>
> Mathieu
