Kafka: Offset commit (manual) failed

[KonstantinCodes]

My consumer always fails to commit the offset:

%4|1560006005.731|COMMITFAIL|rdkafka#consumer-20| [thrd:main]: Offset commit (manual) failed for 1/1 partition(s): Broker: Unknown member: commands[0]@1(Broker: Unknown member)

Here is my Config:

        consumer:
            dsn: "..."
            global:
                ### Make sure this is unique for each application / consumer group and does not change
                ### Otherwise, Kafka won't be able to track your last offset and will always start according to
                ### `auto.offset.reset` setting.
                ### See Kafka documentation regarding `group.id` property if you want to know more
                group.id: 'backend'
                metadata.broker.list: '...'
                security.protocol: sasl_ssl
                ssl.ca.location: '%kernel.project_dir%/config/kafka/ca.pem'
                sasl.username: '...'
                sasl.password: '...'
                sasl.mechanisms: "SCRAM-SHA-256"
                max.poll.interval.ms: '25000'
                enable.auto.commit: 'false'
                enable.auto.offset.store: 'false'
            topic:
                auto.offset.reset: beginning
            ### Commit async is true by default since version 0.9.9.
            ### It is suggested to set it to true in earlier versions since otherwise consumers become extremely slow,
            ### waiting for offset to be stored on Kafka before continuing.
            commit_async: true

Any ideas, why this is happening?

Tested with both heroku Kafka & CloudKarafka.

[Steveb-p]

What offset are you trying to commit?

I've looked up some issues in librdkafka and I've found:

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

Show sample code if possible. I've tried to see what the issue is, but it seems it's something between librdkafka & Kafka broker itself.

[KonstantinCodes]

@Steveb-p Hi, progress has been slow, sorry for late answer.

I published it here: https://github.com/KonstantinCodes/messenger-kafka
I tried to make it as basic as possible and it works for me. In the bottom of the readme, I put the exact config which I'm using. And it runs. I updated the Ticket with the enqueue config, which produced the error to commit.

Hopefully it helps!

[stale]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.

[marcoreni]

@KonstantinCodes @Steveb-p seeing the same issue after upgrading to Symfony 4.3 (and sroze/enqueue-messenger-transport 0.3.2). Everything was working fine before (enqueue 0.9.12, symfony 4.2.*, sroze/enqueue-messenger-transport 0.2.2) .

Maybe this is due to sroze/messenger-enqueue-transport#75 ?

[Steveb-p]

@marcoreni yes, you should upgrade to at least 0.3.0 for sroze/enqueue-messenger-transport. Unfortunately composer does not allow to retroactively mark packages as conflicting as far as I know.

[marcoreni]

This issue came up after upgrading to enqueue-messenger-transport 0.3.2, and the transport is currently unusable (after COMMITFAIL event, the consumer is not able to consume other messages and needs to be restarted).

My configuration is similar as the one in the OP, therefore we're in the case of "automatic async commit" (where I assume offsets are correctly handled)

[Steveb-p]

@marcoreni I'll have a look at it then in the evening. If I'll not be able to figure out why it happens, I might ask folks on phprdkafka. They've made a couple new changes recently as well.

[Steveb-p]

@marcoreni We will probably need a small reproduction sample.

You have upgraded enqueue to the latest version I assume? And you're not using enqueue-kafka directly, but rather through messenger adapter only?

[marcoreni]

https://github.com/marcoreni/enqueue-kafka-894-repro

Here you go. I tried to keep it as simple as possible, I just added two docker containers to make testing easier.
It includes a command to dispatch the message, and a script to dispatch a new message every 5 seconds.

My latest test was able to handle two messages (even if the COMMITFAIL was returned even for the first message:

php_1    | 2019-09-03T09:03:18+00:00 [info] Received message App\Message\TestMessage
php_1    | 2019-09-03T09:03:18+00:00 [info] Received a new message.
php_1    | 2019-09-03T09:03:18+00:00 [info] Message App\Message\TestMessage handled by App\MessageHandler\TestMessageHandler::__invoke
php_1    | 2019-09-03T09:03:18+00:00 [info] App\Message\TestMessage was handled successfully (acknowledging to transport).
php_1    | %4|1567501400.934|COMMITFAIL|rdkafka#consumer-2| [thrd:main]: Offset commit (manual) failed for 1/1 partition(s): Broker: Unknown member: test[0]@4(Broker: Unknown member)
php_1    | 2019-09-03T09:03:24+00:00 [info] Received message App\Message\TestMessage
php_1    | 2019-09-03T09:03:24+00:00 [info] Received a new message.
php_1    | 2019-09-03T09:03:24+00:00 [info] Message App\Message\TestMessage handled by App\MessageHandler\TestMessageHandler::__invoke
php_1    | 2019-09-03T09:03:24+00:00 [info] App\Message\TestMessage was handled successfully (acknowledging to transport).
php_1    | %4|1567501405.934|COMMITFAIL|rdkafka#consumer-4| [thrd:main]: Offset commit (manual) failed for 1/1 partition(s): Broker: Unknown member: test[0]@5(Broker: Unknown member)

Since it's a newly created project I'm using latest version for each package.

(BTW, with this same repository you are also able to reproduce #893. You simply have to stop dispatching messages and wait :) )

[Steveb-p]

@marcoreni Thanks, I'll look into it when I'm free from work :)

[KonstantinCodes]

@KonstantinCodes @Steveb-p seeing the same issue after upgrading to Symfony 4.3 (and sroze/enqueue-messenger-transport 0.3.2). Everything was working fine before (enqueue 0.9.12, symfony 4.2.*, sroze/enqueue-messenger-transport 0.2.2) .

Maybe this is due to sroze/messenger-enqueue-transport#75 ?

Hi, the two issues #893 and #894 already existed before I made the PR. The PR basically just fixed the issue that the offset value was getting lost.

[Steveb-p]

@marcoreni just to let you know I'll have difficulty having a closer look on it before saturday, 07.09. I'm keeping it in the back of my head.

[dirk39]

Hi @Steveb-p , I noticed when testing the application of @marcoreni that when messenger execute the ack, symfony use the QueueInteropTransport::ack that call QueueInteropTransport::getConsumer(). This method call RdKafkaContext::createConsumer that will never check if consumer already exists and create new one. When new consumer try to submit the offset kafka raise the error.
I think is necessary to store created Consumers and return it instead of generating new ones every time. what do you think?

[Slavenin]

Any updates? I have this issue. What composer configuration is work?
I am use
"sroze/messenger-enqueue-transport": "^0.3.2"
"enqueue/rdkafka": "^0.9.12"
but commit not work.

[dirk39]

Hi, the new release with bugfix has not been released yet. In our project, we've changed composer.json to dev-master

[marcoreni]

@Slavenin @dirk39 Fix has been released in 0.9.14.

[stale]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.

[marcoreni]

@Steveb-p This should be closed, it's resolved.

[Steveb-p]

Whoopsie, sorry @marcoreni 😄