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

self,
        topics: str | Sequence[str],
        kafka_config_id: str = "kafka_default",
        apply_function: Callable[..., Any] | str | None = None,
        apply_function_batch: Callable[..., Any] | str | None = None,
        apply_function_args: Sequence[Any] | None = None,
        apply_function_kwargs: dict[Any, Any] | None = None,
        commit_cadence: str | None = "end_of_operator",
        max_messages: int | None = None,
        max_batch_size: int = 1000,
        poll_timeout: float = 60,
        **kwargs: Any,
    ) -> None:
        super().__init__(**kwargs)
        self.topics = topics
        self.apply_function = apply_function
        self.apply_function_batch = apply_function_batch
        self.apply_function_args = apply_function_args or ()
        self.apply_function_kwargs = apply_function_kwargs or {}
        self.kafka_config_id = kafka_config_id
        self.commit_cadence = commit_cadence
        self.max_messages = max_messages or True
        self.max_batch_size = max_batch_size
        self.poll_timeout = poll_timeout
        if self.max_messages is True:
            self.read_to_end = True
        else:
            self.read_to_end = False