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

if self.commit_cadence not in VALID_COMMIT_CADENCE:
            raise AirflowException(
                f"commit_cadence must be one of {VALID_COMMIT_CADENCE}. Got {self.commit_cadence}"
            )
        if self.max_messages and self.max_batch_size > self.max_messages:
            self.log.warning(
                "max_batch_size (%s) > max_messages (%s). Setting max_messages to %s ",
                self.max_batch_size,
                self.max_messages,
                self.max_batch_size,
            )
        if self.commit_cadence == "never":
            self.commit_cadence = None
        if apply_function and apply_function_batch:
            raise AirflowException(
                "One of apply_function or apply_function_batch must be supplied, not both."
            )
    def execute(self, context) -> Any:
        consumer = KafkaConsumerHook(topics=self.topics, kafka_config_id=self.kafka_config_id).get_consumer()
        if isinstance(self.apply_function, str):
            self.apply_function = import_string(self.apply_function)
        if isinstance(self.apply_function_batch, str):
            self.apply_function_batch = import_string(self.apply_function_batch)
        if self.apply_function:
            apply_callable = partial(
                self.apply_function, *self.apply_function_args, **self.apply_function_kwargs  
            )
        if self.apply_function_batch:
            apply_callable = partial(
                self.apply_function_batch,