commit offset failed with No offset stored

[joukosusi]

Description

I create a new topic and try to commit offset on it with three different values: -2, -1, 99, but only 99 worked, others occurred 'Commit failed: Local: No offset stored'。I'm confused on it, could someone help me on this?

confluent-kafka-python: ('0.11.0', 720896)
libversion: ('0.11.3', 721919)
kafka : 0.10.2.1

log:
RESET offset to -2 for topic topic.sucl.new(0) commit offsets: [TopicPartition{topic=topic.sucl.new,partition=0,offset=-2,error=None}] COMMIT: commit offset failed with message: _NO_OFFSET(-168) >> Commit failed: Local: No offset stored
RESET offset to -1 for topic topic.sucl.new(0) commit offsets: [TopicPartition{topic=topic.sucl.new,partition=0,offset=-1,error=None}] COMMIT: commit offset failed with message: _NO_OFFSET(-168) >> Commit failed: Local: No offset stored
RESET offset to 99 for topic topic.sucl.new(0) commit offsets: [TopicPartition{topic=topic.sucl.new,partition=0,offset=99,error=None}]

How to reproduce

`

client = Test(
                hosts='xxx:9093',
                ssl_path=xxxx,
                passwd=None,
                logger=log.LogHandler('test', path='.', log_level='DEBUG').logger,
               new_version=True)

client.reset_offset({'topic.sucl.new': -2})
client.reset_offset({'topic.sucl.new': -1})
client.reset_offset({'topic.sucl.new': 99})

`

definition of class Test
`
class Test():

def __init__(self, hosts, ssl_path, passwd, new_version, logger):
    self._cfg = {
        'group.id': 'group.sucl',
        'bootstrap.servers': hosts,
        'log.connection.close': False,
        'security.protocol': 'ssl' if ssl_path else 'plaintext',
        'ssl.key.location': os.path.join(ssl_path, 'client.key.unsecure') if ssl_path else None,
        'ssl.ca.location': os.path.join(ssl_path, 'ca-cert') if ssl_path else None,
        'ssl.certificate.location': os.path.join(ssl_path, 'client.crt') if ssl_path else None,
        'ssl.key.password': passwd,
        'api.version.request': True if new_version else False,
        'broker.version.fallback': '0.10.2.1' if new_version else '0.9.0',
        'log_level': 6,
        'message.max.bytes': 1000000,
        'log.connection.close': True if new_version else False,
        # 'debug':'generic,broker,topic,metadata',
        'enable.auto.commit': False,
        'auto.commit.interval.ms': 1000,
        'fetch.min.bytes': 1024 * 1024,
        'fetch.wait.max.ms': 1000,
        'fetch.message.max.bytes': 1048576,
        'default.topic.config': {
            'auto.offset.reset': 'smallest',
        },
    }
    self._consumer = Consumer(**self._cfg)
    self._logger = logger

def reset_offset(self, offsets):
    def get_value(para):
        if para in ('min', 'max'):
            return confluent_kafka.OFFSET_BEGINNING if para == 'min' \
                                                    else confluent_kafka.OFFSET_END
        elif isinstance(para, int):
            return para

        self._logger.warning('unknown reset value: {}, will not reset offset'.format(para))
        return None

    if isinstance(offsets, dict):
        assigns = [TopicPartition(_topic, 0, get_value(_offset)) for _topic, _offset in offsets.items()
                   if get_value(_offset)]
    elif isinstance(offsets, list):
        assigns = [TopicPartition(_item[0], _item[1], get_value(_item[2])) for _item in offsets
                   if get_value(_item[2])]
    else:
        self._logger.warning('unknown type: {} for param[offsets], will not reset offset'.format(type(offsets).__name__))
        return None

    [self._logger.warning('RESET offset to {0} for topic {1}({2})'.format(_t.offset, _t.topic, _t.partition))
     for _t in assigns]

    return self.commit(assigns, if_format=True)

def commit(self, offsets=None, if_format=False):
    if if_format:
        topics = offsets
    else:
        src = self._end_offset if offsets is None else offsets
        topics = [TopicPartition(i, k, m) for i, j in src.items() for k, m in j.items()]

    for _ in range(3):
        try:
            self._consumer.commit(offsets=topics, async=False)
            return True
        except (confluent_kafka.KafkaException,) as e:
            _exp_name, exp_code = e.args[0].name(), e.args[0].code()
            if int(exp_code) == 27:  # REBALANCE_IN_PROGRESS
                self._logger.warning('COMMIT: kafka server is now in rebalancing, will retry...')
                time.sleep(1)
                continue
            self._logger.error('COMMIT: commit offset failed with message: {0}({1}) >> {2}'.format(
                e.args[0].name(), e.args[0].code(), e.args[0].str()
            ))
            return False

    self._logger.error('COMMIT: commit failed after 3 times retry')
    return False

`

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]

No offsets stored means the consumer did not see a point in committing any offsets, which has two reasons:

• there are no commits to commit, typically for auto-commit or argument-less commits when there is no current assignment or no messages have been consumed - thus no offsets to commit
• the requested offsets to be committed are already committed.

You can generally handle the NO_OFFSETS error as a success-indicator, the offsets are already committed.

[joukosusi]

Magnus Edenhill (@edenhill) I think two points you mentioned do not exists in my situation(set autocommit False , manually commit TopicPartition info, and only commit once only)... it failed only when offset was beginning(-1), and end(-2)

[edenhill]

Are you trying to commit logical offsets (<0) ?
Only absolute offsets (>= 0) will be committed.

[joukosusi]

Magnus Edenhill (@edenhill) yeah, i'm trying commit logical offsets(OFFSET_BEGINNING and OFFSET_END)

[edenhill]

You can only commit real absolute offsets, not logical.

[joukosusi]

Magnus Edenhill (@edenhill) Thanks! will logical commit support in the future??

[edenhill]

Can you describe your use-case?

[joukosusi]

Magnus Edenhill (@edenhill) sorry for the delay, i just want to reset offset(beginning, end or any absolute offsets) by using one way 。 so checked document and think commit should work and then test like above..

[edenhill]

What you need to do is look up the beginning/end offsets manually using query_watermark_offsets() and then committing those absolute offset(s).

[joukosusi]

Magnus Edenhill (@edenhill) That works! Thanks for your help!!

[VladStatnyk]

Magnus Edenhill (@edenhill) You previously wrote:

No offsets stored means the consumer did not see a point in committing any offsets, which has two reasons:

• there are no commits to commit, typically for auto-commit or argument-less commits when there is no current assignment or no messages have been consumed - thus no offsets to commit

• the requested offsets to be committed are already committed.

We have consumer application, which is working with 'enable.auto.commit': False. Only once we received message from kafka - we are doing commit. Sometimes we are getting Local: No offset stored. We also have retries on exception from commit method. Do you know the possible reasons for getting this exception?

As I understood, if we received already message - we have something to commit, so it's not matching with your first point.

[danigosa]

Magnus Edenhill (@edenhill) You previously wrote:

No offsets stored means the consumer did not see a point in committing any offsets, which has two reasons:

• there are no commits to commit, typically for auto-commit or argument-less commits when there is no current assignment or no messages have been consumed - thus no offsets to commit

• the requested offsets to be committed are already committed.

We have consumer application, which is working with 'enable.auto.commit': False. Only once we received message from kafka - we are doing commit. Sometimes we are getting Local: No offset stored. We also have retries on exception from commit method. Do you know the possible reasons for getting this exception?

As I understood, if we received already message - we have something to commit, so it's not matching with your first point.

It happened for us when using it as no message passed for a bunch of messages, or empty messages consumed with method consume. In any case using last message from the bulk worked consumer.commit(message=msgs[-1], asynchronous=False)

We also assumed that if you consume n messages and the list is not empty, the commit(consumer.commit(asynchronous=False) would work, but even asynchronous=True with callback spits some Local: No offset stored

[eturgeon123]

is there a clear comprehension of what is happening, in the end ? I just experienced the same problem in my code. Auto commit is set to false. I read many records, then commit. I get this error :

Confluent.Kafka.KafkaException: Local: No offset stored
at Confluent.Kafka.Impl.SafeKafkaHandle.Commit(IEnumerable`1 offsets)

most of the time it works though.