Bug in Apache Kafka Provider Consumer Operator

[ahipp13]

Apache Airflow version

2.5.3

What happened

Ran the apache-airflow-providers-apache-kafka version 1.1.2 and got this error:

What you think should happen instead

I was not at the end of my topic and I did not set a max messages so this should have processed another 1000 messages

How to reproduce

Run the provider on a topic that has more than 1000 messages

Operating System

PRETTY_NAME="Debian GNU/Linux 11 (bullseye)" NAME="Debian GNU/Linux" VERSION_ID="11" VERSION="11 (bullseye)" VERSION_CODENAME=bullseye ID=debian HOME_URL="https://www.debian.org/" SUPPORT_URL="https://www.debian.org/support" BUG_REPORT_URL="https://bugs.debian.org/"

Versions of Apache Airflow Providers

apache-airflow-providers-apache-kafka version 1.1.2

Deployment

Official Apache Airflow Helm Chart

Deployment details

Deploying onto Kubernetes

Anything else

This problem occurs every time I run it. Looks like the problem is in "consume.py" as I am using the ConsumeFromTopicOperator. Debugging a little, it is looking like the first 1000 messages it runs fine, but then the second time through for some reason it is setting batch_size wrong:

I think this problems stems from 2 lines. This line: messages_left -= len(msgs) sets the message to -999, but then this line: batch_size = self.max_batch_size if messages_left > self.max_batch_size else messages_left will pick the messages_left which is set to -999.

I believe the fix here would be to change one of the two lines, but from looking at the logic I think the correct change would either be to change to messages_left = len(msgs) or messages_left += len(msgs)

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[eladkal]

cc @dylanbstorey

[aritra24]

Can I take this up? I'd like to make a fix for this.

[hussein-awala]

@aritra24 I assigned you, feel free to open a PR to fix the issue.

[ahipp13]

Can I take this up? I'd like to make a fix for this.

@aritra24 make sure max_messages also works after your change, looks like after mine max messages stopped working so looks like the whole code will have to be looked at.

[aritra24]

Thank for the heads-up @ahipp13 will take care of that