Consumer Keeps Resets to -1001 and difference between topic offset and consumer offset?

[rohitgcs]

Description

So I'm currently troubleshooting my consumer and trying different things with partitions. I have 3 partitions and a consumer for each partition.
I initially had an issue where the commit offset would be -1001 but I figured that was only because of the timeout. So I put in code to reset it to 0 if <0, but now everytime I rerun the consumer it always returns -1001 as my offset.
Is there a way to find the latest commit of a particular topic partition?
And also, what is the difference between topic and consumer offset?

Thanks in advance

How to reproduce

consumer = Consumer({'bootstrap.servers': KAFKA_BROKER,
... 'group.id': 'maxwellConsumer','debug':'broker,fetch',
... 'enable.auto.commit': True,
... 'default.topic.config': {'auto.offset.reset': 'latest'}})
%7|1513895424.653|BRKMAIN|rdkafka#consumer-2| [thrd::0/internal]: :0/internal: Enter main broker thread
%7|1513895424.653|STATE|rdkafka#consumer-2| [thrd::0/internal]: :0/internal: Broker changed state INIT -> UP
%7|1513895424.654|BROKER|rdkafka#consumer-2| [thrd:app]: 172.31.230.234:9092/bootstrap: Added new broker with NodeId -1
%7|1513895424.654|BRKMAIN|rdkafka#consumer-2| [thrd:172.31.230.234:9092/bootstrap]: 172.31.230.234:9092/bootstrap: Enter main broker thread
%7|1513895424.654|CONNECT|rdkafka#consumer-2| [thrd:172.31.230.234:9092/bootstrap]: 172.31.230.234:9092/bootstrap: broker in state INIT connecting
%7|1513895424.654|CONNECT|rdkafka#consumer-2| [thrd:172.31.230.234:9092/bootstrap]: 172.31.230.234:9092/bootstrap: Connecting to ipv4#172.31.230.234:9092 (plaintext) with socket 7
%7|1513895424.654|STATE|rdkafka#consumer-2| [thrd:172.31.230.234:9092/bootstrap]: 172.31.230.234:9092/bootstrap: Broker changed state INIT -> CONNECT
%7|1513895424.654|CONNECT|rdkafka#consumer-2| [thrd:172.31.230.234:9092/bootstrap]: 172.31.230.234:9092/bootstrap: Connected to ipv4#172.31.230.234:9092
%7|1513895424.654|CONNECTED|rdkafka#consumer-2| [thrd:172.31.230.234:9092/bootstrap]: 172.31.230.234:9092/bootstrap: Connected (#1)
%7|1513895424.654|FEATURE|rdkafka#consumer-2| [thrd:172.31.230.234:9092/bootstrap]: 172.31.230.234:9092/bootstrap: Updated enabled protocol features +ApiVersion to ApiVersion
%7|1513895424.654|STATE|rdkafka#consumer-2| [thrd:172.31.230.234:9092/bootstrap]: 172.31.230.234:9092/bootstrap: Broker changed state CONNECT -> APIVERSION_QUERY
%7|1513895424.655|FEATURE|rdkafka#consumer-2| [thrd:172.31.230.234:9092/bootstrap]: 172.31.230.234:9092/bootstrap: Updated enabled protocol features to MsgVer1,ApiVersion,BrokerBalancedConsumer,ThrottleTime,Sasl,SaslHandshake,BrokerGroupCoordinator,LZ4,OffsetTime,MsgVer2
%7|1513895424.655|STATE|rdkafka#consumer-2| [thrd:172.31.230.234:9092/bootstrap]: 172.31.230.234:9092/bootstrap: Broker changed state APIVERSION_QUERY -> UP
%7|1513895424.655|BROKER|rdkafka#consumer-2| [thrd:main]: 192.23.213.130:9092/2: Added new broker with NodeId 2
%7|1513895424.655|BRKMAIN|rdkafka#consumer-2| [thrd:192.23.213.130:9092/2]: 192.23.213.130:9092/2: Enter main broker thread
%7|1513895424.655|CONNECT|rdkafka#consumer-2| [thrd:192.23.213.130:9092/2]: 192.23.213.130:9092/2: broker in state INIT connecting
%7|1513895424.655|BROKER|rdkafka#consumer-2| [thrd:main]: 172.31.230.155:9092/1: Added new broker with NodeId 1
%7|1513895424.655|CONNECT|rdkafka#consumer-2| [thrd:192.23.213.130:9092/2]: 192.23.213.130:9092/2: Connecting to ipv4#192.23.213.130:9092 (plaintext) with socket 10
%7|1513895424.655|CLUSTERID|rdkafka#consumer-2| [thrd:main]: 172.31.230.234:9092/bootstrap: ClusterId update "" -> "LlhmfovJSe-sOmvvhbrI7w"
%7|1513895424.655|UPDATE|rdkafka#consumer-2| [thrd:172.31.230.234:9092/bootstrap]: 172.31.230.234:9092/bootstrap: NodeId changed from -1 to 0
%7|1513895424.655|UPDATE|rdkafka#consumer-2| [thrd:172.31.230.234:9092/bootstrap]: 172.31.230.234:9092/0: Name changed from 172.31.230.234:9092/bootstrap to 172.31.230.234:9092/0
%7|1513895424.655|LEADER|rdkafka#consumer-2| [thrd:172.31.230.234:9092/bootstrap]: 172.31.230.234:9092/0: Mapped 0 partition(s) to broker
%7|1513895424.655|STATE|rdkafka#consumer-2| [thrd:172.31.230.234:9092/bootstrap]: 172.31.230.234:9092/0: Broker changed state UP -> UPDATE
%7|1513895424.655|STATE|rdkafka#consumer-2| [thrd:192.23.213.130:9092/2]: 192.23.213.130:9092/2: Broker changed state INIT -> CONNECT
%7|1513895424.655|BRKMAIN|rdkafka#consumer-2| [thrd:172.31.230.155:9092/1]: 172.31.230.155:9092/1: Enter main broker thread
%7|1513895424.655|CONNECT|rdkafka#consumer-2| [thrd:172.31.230.155:9092/1]: 172.31.230.155:9092/1: broker in state INIT connecting
%7|1513895424.655|CONNECT|rdkafka#consumer-2| [thrd:172.31.230.155:9092/1]: 172.31.230.155:9092/1: Connecting to ipv4#172.31.230.155:9092 (plaintext) with socket 13
%7|1513895424.656|STATE|rdkafka#consumer-2| [thrd:172.31.230.155:9092/1]: 172.31.230.155:9092/1: Broker changed state INIT -> CONNECT
%7|1513895424.656|STATE|rdkafka#consumer-2| [thrd:172.31.230.234:9092/bootstrap]: 172.31.230.234:9092/0: Broker changed state UPDATE -> UP
%7|1513895424.656|CONNECT|rdkafka#consumer-2| [thrd:192.23.213.130:9092/2]: 192.23.213.130:9092/2: Connected to ipv4#192.23.213.130:9092
%7|1513895424.656|CONNECTED|rdkafka#consumer-2| [thrd:192.23.213.130:9092/2]: 192.23.213.130:9092/2: Connected (#1)
%7|1513895424.656|FEATURE|rdkafka#consumer-2| [thrd:192.23.213.130:9092/2]: 192.23.213.130:9092/2: Updated enabled protocol features +ApiVersion to ApiVersion
%7|1513895424.656|STATE|rdkafka#consumer-2| [thrd:192.23.213.130:9092/2]: 192.23.213.130:9092/2: Broker changed state CONNECT -> APIVERSION_QUERY
%7|1513895424.656|CONNECT|rdkafka#consumer-2| [thrd:172.31.230.155:9092/1]: 172.31.230.155:9092/1: Connected to ipv4#172.31.230.155:9092
%7|1513895424.656|CONNECTED|rdkafka#consumer-2| [thrd:172.31.230.155:9092/1]: 172.31.230.155:9092/1: Connected (#1)
%7|1513895424.656|FEATURE|rdkafka#consumer-2| [thrd:172.31.230.155:9092/1]: 172.31.230.155:9092/1: Updated enabled protocol features +ApiVersion to ApiVersion
%7|1513895424.656|STATE|rdkafka#consumer-2| [thrd:172.31.230.155:9092/1]: 172.31.230.155:9092/1: Broker changed state CONNECT -> APIVERSION_QUERY
%7|1513895424.656|FEATURE|rdkafka#consumer-2| [thrd:192.23.213.130:9092/2]: 192.23.213.130:9092/2: Updated enabled protocol features to MsgVer1,ApiVersion,BrokerBalancedConsumer,ThrottleTime,Sasl,SaslHandshake,BrokerGroupCoordinator,LZ4,OffsetTime,MsgVer2
%7|1513895424.656|STATE|rdkafka#consumer-2| [thrd:192.23.213.130:9092/2]: 192.23.213.130:9092/2: Broker changed state APIVERSION_QUERY -> UP
%7|1513895424.656|FEATURE|rdkafka#consumer-2| [thrd:172.31.230.155:9092/1]: 172.31.230.155:9092/1: Updated enabled protocol features to MsgVer1,ApiVersion,BrokerBalancedConsumer,ThrottleTime,Sasl,SaslHandshake,BrokerGroupCoordinator,LZ4,OffsetTime,MsgVer2
%7|1513895424.656|STATE|rdkafka#consumer-2| [thrd:172.31.230.155:9092/1]: 172.31.230.155:9092/1: Broker changed state APIVERSION_QUERY -> UP

topic = TopicPartition('elastic',50)
print "Topic offset",topic.offset
Topic offset -1001
consumer.assign([topic])
%7|1513895439.655|TOPBRK|rdkafka#consumer-2| [thrd:172.31.230.234:9092/bootstrap]: 172.31.230.234:9092/0: Topic elastic [2]: joining broker (rktp 0x7f9ffc004e70)
%7|1513895439.655|TOPBRK|rdkafka#consumer-2| [thrd:172.31.230.155:9092/1]: 172.31.230.155:9092/1: Topic elastic [0]: joining broker (rktp 0x7f9ffc0045f0)
%7|1513895439.655|TOPBRK|rdkafka#consumer-2| [thrd:192.23.213.130:9092/2]: 192.23.213.130:9092/2: Topic elastic [1]: joining broker (rktp 0x7f9ffc0049d0)

confluent_kafka.version() - ('0.11.0', 720896)
confluent_kafka.libversion() - ('0.11.0', 721151)
Broker version 0.9.0.1
OS: ubuntu

Checklist

Please provide the following information:

• confluent-kafka-python and librdkafka version (confluent_kafka.version() and confluent_kafka.libversion()):
• Apache Kafka broker version:
• Client configuration: {...}
• Operating system:
• Provide client logs (with 'debug': '..' as necessary)
• Provide broker log excerpts
• Critical issue

[edenhill]

Offset -1001 is a special value meaning Invalid or Default, depending on context.
If you pass a TopicPartition list to assign() with offset set to -1001 it will attempt to fetch committed offsets from the broker, and if that fails revert to auto.offset.reset.

If you want to manually specify the starting offset you can set the offset of the TopicPartition to an offset to your liking.

You can use position() to get the current consume position for the given partitions, and committed() to get the comitted offsets. Use assignment() to get the currently assigned partition set.

[rohitgcs]

@edenhill everytime I stop the consumer the offest resets to -1001
So currently I have it reset to 0, but that's not helping me. I basically need it to reset to the last commit it had (but that prints out -1001 most of the time.

topic = TopicPartition(FROM_TOPIC,i)
print "Topic offset",topic.offset
consumer.assign([topic])
print "Getting Position"
position = consumer.position([topic])
if position[0].offset<0:
position[0].offset=0 #Reset Offset

[treziac]

Can you provide your config and some code?
In order to resume from previous offset, you have to use the same consumer group id when restarting a consumer, and commit offset regularly (client does it every 5s by default, but you should adapt it to your use case)

When you say you have a consumer for each partition, are you manually assigning the partition, or do you have three consumer in a same consumer groups which share the offset? In your case, you should use the same group id for each consumer, as you seem to want them to read all data from topics. Your call if you use subscribe (auto repartition) or assign (you decide which partition each consumer read).

If you assign the consumer to the partition, you should let the invalid offset for the assignment- this will use the stored offset for the consumer group id.

With some code it will be easier to help you :)

[rohitgcs]

@treziac @edenhill this is how I set up my consumers. I do use the same groupid and I assign them manually.

def launchConsumer(i):
    consumer = Consumer({'bootstrap.servers': KAFKA_BROKER, \
    'group.id': 'testConsumer', \
    'enable.auto.commit': True,\
    'default.topic.config': {'auto.offset.reset': 'latest'}})
    topic = TopicPartition(FROM_TOPIC,i)
    print "Topic offset",topic.offset
    consumer.assign([topic])
    print "Getting Position"
    position = consumer.position([topic])
    if position[0].offset<0:
        position[0].offset=0 #Reset Offset
    print "Partition", i, ", Position: ", position
    hasData = True
    try:
        while hasData:
            print "Running..."
            message = consumer.poll()
            if message:
                logger.debug("Consumer: Data Retrieved")
                print "================================================"
                print dumps(loads(message.value()), indent=4)
				msg = loads(message.value())
	            if msg['type']=='insert':
	            	# Do Something
                print "================================================"
    except KeyboardInterrupt:
        sys.stderr.write('%% Aborted by user\n')
        logger.debug('%% Aborted by user\n')
        c.close()
    c.close()

if __name__ == "__main__":
    threads = list()
    topic = FROM_TOPIC
	threadCount = int(sys.argv[1])
    for i in range(threadCount):
        t = threading.Thread(target=launchConsumer, args=(i,)) # pass in the callable
        threads.append(t)
        print('Starting Thread {}'.format(i))
        t.start()
    for i, t in enumerate(threads):
        t.join()
        print('Thread {} Stopped'.format(i))

[edenhill]

   consumer.assign([topic])
    print "Getting Position"
    position = consumer.position([topic])

Since the client is asynchronous internally and it takes some time to set up connections, query leaders, and start fetching, it is very likely that calling position() right after assign() will not return an actual offset.

What I think you are after is actually the committed offset, for which you would use committed() (and you can do so right after assign() since it will block while waiting for the broker to respond).
position() is instantaneous and returns the current fetch position, calling it before the fetching has started will give you -1001.

Unrelated, check message for None: if message is not None:, since a zero length message (e.g., a delete) will be false otherwhise.

[rohitgcs]

@edenhill so to your point I need to wait before fetching the offset? so a time.sleep() should help?
Something like

def launchConsumer(i):
    consumer = Consumer({'bootstrap.servers': KAFKA_BROKER, \
    'group.id': 'testConsumer', \
    'enable.auto.commit': True,\
    'default.topic.config': {'auto.offset.reset': 'latest'}})
    topic = TopicPartition(FROM_TOPIC,i)
    print "Topic offset",topic.offset
    consumer.assign([topic])
    print "Getting Position"
    time.sleep(5)
    position = consumer.position([topic])
    if position[0].offset<0:
        position[0].offset=0 #Reset Offset
    print "Partition", i, ", Position: ", position

Is that right?

[edenhill]

What is it that you are trying to achieve by getting the fetch position?

[rohitgcs]

@edenhill I'm trying what you mentioned, if for some reason if the consumer crashes and dies - I would like to start this script again. This way, when the consumers launches and I start assigning partitions to each consumer, I want to get the last offset position and continue. I check for position<0 because of the -1001 so if there's a better way you could reconstruct my code?

[edenhill]

The client already has built-in support for this by two means:

• resuming from committed offsets
• if no committed offsets are available or they are out of date: revert to the auto.offset.reset policy.

Which means you only need to decide if you want to start reading the oldest (auto.offset.reset=beginning) or latest (auto.offset.reset=end) if no committed offsets are available, and you do this by configuration alone.

[rohitgcs]

@edenhill okay, will take a look.
How is this different from looking for get_watermark_offsets() for a partition? Where would I need to use watermark offset as supposed to a general offset.

[edenhill]

It is not different, it is just handled automatically in the client itself, so there is typically no reason to implement the same logic in the application.

[rohitgcs]

cool, let me try your changes and get back.

[rohitgcs]

@edenhill I removed setting offset to 0 and I noticed form the fetch logs that the consumer gets stuck on the latest offset and doesn't cover the lag.

%7|1514946451.579|FETCH|rdkafka#consumer-28| [thrd:172.32.231.127:9092/2]: 172.32.231.127:9092/2: Fetch topic qa_gcdb_shipment_queue [26] at offset 2413 (v2)