### Apache Airflow version

2.5.3

### What happened

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

![image](https://github.com/apache/airflow/assets/118911990/bbb24811-8770-4210-b701-b4aa696c0fdf)


### 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:

![image](https://github.com/apache/airflow/assets/118911990/1c571a44-d1be-4eeb-b87b-dc4db66d120b)

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?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

:param apply_function_args: Additional arguments that should be applied to the callable, defaults to None
    :param apply_function_kwargs: Additional key word arguments that should be applied to the callable
        defaults to None
    :param commit_cadence: When consumers should commit offsets ("never", "end_of_batch","end_of_operator"),
        defaults to "end_of_operator";
        if end_of_operator, the commit() is called based on the max_messages arg. Commits are made after the
        operator has processed the apply_function method for the maximum messages in the operator.
        if end_of_batch, the commit() is called based on the max_batch_size arg. Commits are made after each
        batch has processed by the apply_function method for all messages in the batch.
        if never,  close() is called without calling the commit() method.
    :param max_messages: The maximum total number of messages an operator should read from Kafka,
        defaults to None implying read to the end of the topic.
    :param max_batch_size: The maximum number of messages a consumer should read when polling,
        defaults to 1000
    :param poll_timeout: How long the Kafka consumer should wait before determining no more messages are
        available, defaults to 60
    .. seealso::
        For more information on how to use this operator, take a look at the guide:
        :ref:`howto/operator:ConsumeFromTopicOperator`
    """
    BLUE = "#ffefeb"
    ui_color = BLUE
    template_fields = (
        "topics",
        "apply_function",
        "apply_function_args",
        "apply_function_kwargs",
        "kafka_config_id",
    )
    def __init__(