Add InteropMessageStamp, to enable ACKs

KonstantinCodes commented 5 years ago

This is a proposed solution for the ACK problem. I hope you find this suggestion useful.

Bug report: #74 #76

Steveb-p commented 5 years ago

@KonstantinCodes PR fails due to PHP-CS.

KonstantinCodes commented 5 years ago

@Steveb-p thanks for the heads up! I added the Symfony copyright

alsar commented 5 years ago

@KonstantinCodes when using your patch I get an error on Kafka: 1560235440.674|COMMITFAIL|rdkafka#consumer-6| [thrd:main]: Offset commit (manual) failed for 1/1 partition(s): Broker: Unknown member: test[0]@9(Broker: Unknown member)

but I don't know if it's related to your patch or something in the 0.3.0 version. On 0.2.2 it was working fine.

KonstantinCodes commented 5 years ago

@alsar This PR seems to work, because it fixed the same issue for the AMQP transport: https://github.com/sroze/messenger-enqueue-transport/issues/76#issuecomment-499877875 And also gets the Kafka consumer one step further in ACKing messages.

I raised this specific Kafka problem in https://github.com/php-enqueue/enqueue-dev/issues/894

I'm not sure, where this is coming from. Debugging was inconclusive, so I tried building a Kafka transport for Symfony Messenger using the basic RdKafka Classes. I published it here: https://github.com/KonstantinCodes/messenger-kafka

I implemented the same ACKing mechanism as is contained in this PR, just not using the enqueue components. And it works fine for me.

So I suspect that the error is inside of the enqueue components. Maybe in the code that sets up the configuration?