InOrderConsumer doesn't work as intended.

[AdrielVelazquez]

💸 TL;DR + Details

The InOrderConsumer Kafka Class makes a lot of assumptions that are False and inefficient with how Kafka Processes Events.

The current logic of using only 1 consumer and commit back the offsets doesn't take into consideration that the upstream messages are sending messages randomly to each partition.

You would need to enable transactions upstream for the producer, as well as delivering messages with a Key. This would guarantee two things.

1. All messages with a given key are guaranteed to land in a single partition.

https://docs.confluent.io/platform/current/clients/producer.html#concepts

The partitioners shipped with Kafka guarantee that all messages with the same non-empty key will be sent to the same partition.

If the key is provided, the partitioner will hash the key with murmur2 algorithm and divide it by the number of partitions. The result is that the same key is always assigned to the same partition.

2. Consumers can switch to reading committed transactions instead of uncommitted transactions.

https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html#isolation-level

📜 Details

So what is this PR doing so differently?

Technically, this keeps the same exact APIs in place, but only solves 1 problem. Stop committing back to Kafka so frequently.

Instead of manually committing every offset which swarms Kafka with requests. Store the offset locally and commit the last max seen offset per partition back to Kafka in the auto.commit frequency.

This keeps the behavior that you're committing every offset, but only delivering it every couple of seconds.

One More Detail

My lilnter went a little crazy when committing this. I can revert that.

[chriskuehl]

This change looks reasonable to me, but I'm not very familiar with Kafka so I can't really provide insight into the decision itself. Could you nominate an additional reviewer (maybe someone from your team) who can double-check that the change looks right before we merge it?

[tomncooper]

The linting changes do make it harder to see what has logically changed. I am assuming the main changes are lines: 476, 477 and 496.

The only comments I have are a few language nits on the doc string and a question over the value of auto.commit.interval.ms. If the consumer crashes then there is a now a chance of duplicate message reads. The worst case being crashing just before the end of an interval in which case you would replay auto.commit.interval.ms worth of messages. Which, for what it is worth, is pretty standard practice for Kafka consumers.

Do we want to surface that, or is auto.commit.interval.ms already set to a low value? Do we want to allow users to tune that?

[skykistler]

Just one clarification to the comment

[AdrielVelazquez]

@tomncooper and @skykistler This is good to review again.

I mentioned the config and the default in the docstring, but I would rather not be explicit on the value.

[AdrielVelazquez]

@chriskuehl Sorry should have tagged you also. But I cleaned up all the linting formatting my setup had.