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

#
#
from __future__ import annotations
from functools import partial
from typing import Any, Callable, Sequence
from airflow.exceptions import AirflowException
from airflow.models import BaseOperator
from airflow.providers.apache.kafka.hooks.consume import KafkaConsumerHook
from airflow.utils.module_loading import import_string
VALID_COMMIT_CADENCE = {"never", "end_of_batch", "end_of_operator"}
class ConsumeFromTopicOperator(BaseOperator):
    """An operator that consumes from Kafka a topic(s) and processing the messages.
    The operator creates a Kafka consumer that reads a batch of messages from the cluster and processes them
    using the user supplied callable function. The consumer will continue to read in batches until it reaches
    the end of the log or reads a maximum number of messages is reached.
    :param kafka_config_id: The connection object to use, defaults to "kafka_default"
    :param topics: A list of topics or regex patterns the consumer should subscribe to.
    :param apply_function: The function that should be applied to fetched one at a time.
        name of dag file executing the function and the function name delimited by a `.`
    :param apply_function_batch: The function that should be applied to a batch of messages fetched. Can not
        be used with `apply_function`. Intended for transactional workloads where an expensive task might
        be called before or after operations on the messages are taken.