### Apache Airflow Provider(s)

amazon

### Versions of Apache Airflow Providers

apache-airflow-providers-amazon==7.1.0

### Apache Airflow version

2.5.1

### Operating System

Ubuntu 18.04

### Deployment

Official Apache Airflow Helm Chart

### Deployment details

_No response_

### What happened

When looking at logs in the UI for a running task when using remote s3 logging, the logs for the task are only uploaded to s3 after the task has completed. The the `S3TaskHandler` falls back to the local logs stored on the worker in that case (by falling back to the `FileTaskHandler` behavior) and prepends the line `*** Falling back to local log` to those logs.

This is mostly fine, but for the new log streaming behavior, this means that `*** Falling back to local log` is returned from `/get_logs_with_metadata` on each call, even if there are no new logs.

### What you think should happen instead

I'd expect the falling back message only to be included in calls with no `log_pos` in the metadata or with a `log_pos` of `0`.

### How to reproduce

Start a task with `logging.remote_logging` set to `True` and `logging.remote_base_log_folder` set to `s3://something` and watch the logs while the task is running. You'll see `*** Falling back to local log` printed every few seconds.

### Anything else

_No response_

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

@conf_vars({("logging", "remote_log_conn_id"): "aws_default"})
    @pytest.fixture(autouse=True)
    def setup_tests(self, create_log_template, tmp_path_factory):
        self.remote_log_base = "s3://bucket/remote/log/location"
        self.remote_log_location = "s3://bucket/remote/log/location/1.log"
        self.remote_log_key = "remote/log/location/1.log"
        self.local_log_location = str(tmp_path_factory.mktemp("local-s3-log-location"))
        create_log_template("{try_number}.log")
        self.s3_task_handler = S3TaskHandler(self.local_log_location, self.remote_log_base)
        
        assert self.s3_task_handler.hook is not None
        date = datetime(2016, 1, 1)
        self.dag = DAG("dag_for_testing_s3_task_handler", start_date=date)
        task = EmptyOperator(task_id="task_for_testing_s3_log_handler", dag=self.dag)
        dag_run = DagRun(dag_id=self.dag.dag_id, execution_date=date, run_id="test", run_type="manual")
        with create_session() as session:
            session.add(dag_run)
            session.commit()
            session.refresh(dag_run)
        self.ti = TaskInstance(task=task, run_id=dag_run.run_id)
        self.ti.dag_run = dag_run
        self.ti.try_number = 1
        self.ti.state = State.RUNNING
        self.conn = boto3.client("s3")
        
        
        moto.moto_api._internal.models.moto_api_backend.reset()
        self.conn.create_bucket(Bucket="bucket")