Expose offsets_for_times consumer method. closes #224

[johnistan]

Expose offsets_for_times consumer method. closes #224

[johnistan]

I have tested this on an actual kafka instance, but there doesn't appear to be integration tests. So i have only checked in the basic smoke test.

[johnistan]

failing on untouched pep8 issues

./confluent_kafka/avro/cached_schema_registry_client.py:171:13: E722 do not use bare except'
./confluent_kafka/avro/cached_schema_registry_client.py:207:13: E722 do not use bare except'
./confluent_kafka/avro/cached_schema_registry_client.py:272:9: E722 do not use bare except'
./confluent_kafka/avro/serializer/message_serializer.py:45:1: E722 do not use bare except'
./confluent_kafka/avro/serializer/message_serializer.py:161:9: E722 do not use bare except'
./confluent_kafka/avro/serializer/message_serializer.py:183:13: E722 do not use bare except'
./examples/consumer.py:62:9: E722 do not use bare except'
./tests/test_threads.py:8:1: E722 do not use bare except'
./tests/avro/mock_registry.py:123:9: E722 do not use bare except'

[johnistan]

Other then the flake8 fails. anything this needs?

[johnistan]

bump. happy to respond to any comments.

[edenhill]

One doc fix needed.
Please also add a call to offsets_for_times() to the consumer integration test, just making sure it returns some valid offsets.

[edenhill]

remove cache mention

[edenhill]

Thank you for this, almost there!

[johnistan]

After some more testing I think i have found an issue with calling offsets_for_time with no timeout. The call just hangs and never returns. I need to dig in more but have you seen behavior like this before from the underlying C function?

[johnistan]

Found some more issues. The first call from a newly created consumer will fail like so:

offsets = c.offsets_for_times(topic_partitions_to_search)
cimpl.KafkaException: KafkaError{code=LEADER_NOT_AVAILABLE,val=5,str="Failed to get offsets: Broker: Leader not available"}

I suspect the same issue a #196 and underlying confluentinc/librdkafka#1373

Adding a timeout causes it to work.

[johnistan]

It could be that my code has an issue but I think there are issues with the underlying C implementation. Without a formal integration setup that spins up a fresh kafka, sharing a reproducible example is a pain but here is a simple script:

This script shows two issues. First calling offset_for_times without a timeout or a -1 timeout, returns a copy of the TopicPartition objects you passed in. Second, calling offset_for_times without a timeout causes any subsequent close() on the consumer to hang.

import confluent_kafka
import uuid

topic_name = 'test-offsets'

conf = {
    'bootstrap.servers': 'localhost:9092',
    'api.version.request': True
}

# Create producer
p = confluent_kafka.Producer(**conf)

for i in range(1000):
    p.produce(topic_name, 'foo', timestamp=123456789000)
p.flush()

c_conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': uuid.uuid1(),
    'session.timeout.ms': 6000,
    'api.version.request': True
}
c = confluent_kafka.Consumer(**c_conf)

topic_partitions_to_search = list(map(lambda p: confluent_kafka.TopicPartition(topic_name, p, 123456788000), range(0, 3)))
print("topic_partitions_to_search: %s" % topic_partitions_to_search)
# hack to make sure consumer has topic/broker metadata
c.subscribe([topic_name])
msg = c.poll()

offsets_with_timeout = c.offsets_for_times(topic_partitions_to_search, timeout=1.0)
offsets_with_neg_1_timeout = c.offsets_for_times(topic_partitions_to_search, timeout=-1)
offsets_without_timeout = c.offsets_for_times(topic_partitions_to_search)

print("offsets_with_timeout: %s" % offsets_with_timeout)
print("offsets_with_neg_1_timeout: %s" % offsets_with_neg_1_timeout)
print("offsets_without_timeout: %s" % offsets_without_timeout)

print("Right before close")
c.close()
print("Right after close")

Outputs:

py27 runtests: commands[1] | python examples/test.py
topic_partitions_to_search: [TopicPartition{topic=test-offsets,partition=0,offset=123456788000,error=None}, TopicPartition{topic=test-offsets,partition=1,offset=123456788000,error=None}, TopicPartition{topic=test-offsets,partition=2,offset=123456788000,error=None}]
offsets_with_timeout: [TopicPartition{topic=test-offsets,partition=0,offset=8557,error=None}, TopicPartition{topic=test-offsets,partition=1,offset=8179,error=None}, TopicPartition{topic=test-offsets,partition=2,offset=9269,error=None}]
offsets_with_neg_1_timeout: [TopicPartition{topic=test-offsets,partition=0,offset=123456788000,error=None}, TopicPartition{topic=test-offsets,partition=1,offset=123456788000,error=None}, TopicPartition{topic=test-offsets,partition=2,offset=123456788000,error=None}]
offsets_without_timeout: [TopicPartition{topic=test-offsets,partition=0,offset=123456788000,error=None}, TopicPartition{topic=test-offsets,partition=1,offset=123456788000,error=None}, TopicPartition{topic=test-offsets,partition=2,offset=123456788000,error=None}]
Right before close
^CERROR: KEYBOARDINTERRUPT
C^CERROR: keyboardinterrupt

[johnistan]

Running locally with these changes to librdkafka: confluentinc/librdkafka#1525 i get good results:

py27 runtests: commands[1] | python examples/test.py
topic_partitions_to_search: [TopicPartition{topic=test-offsets,partition=0,offset=123456788000,error=None}, TopicPartition{topic=test-offsets,partition=1,offset=123456788000,error=None}, TopicPartition{topic=test-offsets,partition=2,offset=123456788000,error=None}]
offsets_with_timeout: [TopicPartition{topic=test-offsets,partition=0,offset=9888,error=None}, TopicPartition{topic=test-offsets,partition=1,offset=9406,error=None}, TopicPartition{topic=test-offsets,partition=2,offset=10711,error=None}]
offsets_with_neg_1_timeout: [TopicPartition{topic=test-offsets,partition=0,offset=9888,error=None}, TopicPartition{topic=test-offsets,partition=1,offset=9406,error=None}, TopicPartition{topic=test-offsets,partition=2,offset=10711,error=None}]
offsets_without_timeout: [TopicPartition{topic=test-offsets,partition=0,offset=9888,error=None}, TopicPartition{topic=test-offsets,partition=1,offset=9406,error=None}, TopicPartition{topic=test-offsets,partition=2,offset=10711,error=None}]
Right before close
Right after close
__________________________________________________________________________________________________________________ summary ___________________________________________________________________________________________________________________
  py27: commands succeeded
  congratulations :)

[matthew-d-jones]

@edenhill I think this should have closed #225 and not #224.

[DestroyerAlpha]

Is there proper documentation for this?

[matthew-d-jones]

@DestroyerAlpha https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#confluent_kafka.Consumer.offsets_for_times