KAFKA-3977: Exception in value_deserializer in KafkaConsumer should be raised to user

[zackdever]

I have an avro deserializer in my KafkaConsumer and have a message that's throwing a SchemaResolutionException. I expect the consumer to crash, but this is executed in a Future as part of _handle_fetch_response, so the error is logged, but not raised because Future.error_on_callbacks = False by default.

It seems like the consumer is infinitely retrying to deserialize this message and spamming the logs. Ideally exceptions caused by deserialization would be raised, or retried a finite set of times before raising.

Example error:

Traceback (most recent call last):
  File "/Users/zdever/.virtualenvs/sod-index/lib/python2.7/site-packages/kafka/future.py", line 79, in _call_backs
    f(value)
  File "/Users/zdever/.virtualenvs/sod-index/lib/python2.7/site-packages/kafka/consumer/fetcher.py", line 760, in _handle_fetch_response
    unpacked = list(self._unpack_message_set(tp, messages))
  File "/Users/zdever/.virtualenvs/sod-index/lib/python2.7/site-packages/kafka/consumer/fetcher.py", line 539, in _unpack_message_set
    tp.topic, msg.value)
  File "/Users/zdever/.virtualenvs/sod-index/lib/python2.7/site-packages/kafka/consumer/fetcher.py", line 570, in _deserialize
    return f(bytes_)
  File "sod/scrapers/curated_playlists.py", line 104, in avro_deserialize
    return _avro_reader.read(avro.io.BinaryDecoder(buf))
  File "/Users/zdever/.virtualenvs/sod-index/lib/python2.7/site-packages/avro/io.py", line 445, in read
    return self.read_data(self.writers_schema, self.readers_schema, decoder)
  File "/Users/zdever/.virtualenvs/sod-index/lib/python2.7/site-packages/avro/io.py", line 490, in read_data
    return self.read_record(writers_schema, readers_schema, decoder)
  File "/Users/zdever/.virtualenvs/sod-index/lib/python2.7/site-packages/avro/io.py", line 690, in read_record
    field_val = self.read_data(field.type, readers_field.type, decoder)
  File "/Users/zdever/.virtualenvs/sod-index/lib/python2.7/site-packages/avro/io.py", line 488, in read_data
    return self.read_union(writers_schema, readers_schema, decoder)
  File "/Users/zdever/.virtualenvs/sod-index/lib/python2.7/site-packages/avro/io.py", line 650, in read_union
    raise SchemaResolutionException(fail_msg, writers_schema, readers_schema)
SchemaResolutionException: Can't access branch index 1499670000000 for union with 2 branches
Writer's Schema: [
  "null",
  "long"
]
Reader's Schema: [
  "null",
  "long"
]```

[dpkp]

Upstream: https://issues.apache.org/jira/browse/KAFKA-3977

[dpkp]

But also -- was this fixed in #1159 ? I can't reproduce this w/ kafka-python 1.3.4

[jeffwidman]

@dpkp what's the status on this? I see:

I can't reproduce this w/ kafka-python 1.3.4

But then you put up PR #1245. Was that just to bring the code path up to date with Java, or were you able to replicate the issue so needed the PR to fix it?

[dpkp]

I believe technically this is fixed in master. PR 1245 brings the code into alignment w/ the official java client. It also makes the stacktrace easier to read (I think) because the exception will be raised in the standard record flow rather than via a network callback.

I'm fine closing now or waiting for the PR to land.