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

mock_open = mock.mock_open()
        with mock.patch("airflow.providers.amazon.aws.log.s3_task_handler.open", mock_open):
            self.s3_task_handler.set_context(self.ti)
        assert self.s3_task_handler.upload_on_close
        mock_open.assert_called_once_with(os.path.join(self.local_log_location, "1.log"), "w")
        mock_open().write.assert_not_called()
    def test_read(self):
        self.conn.put_object(Bucket="bucket", Key=self.remote_log_key, Body=b"Log line\n")
        ti = copy.copy(self.ti)
        ti.state = TaskInstanceState.SUCCESS
        log, metadata = self.s3_task_handler.read(ti)
        actual = log[0][0][-1]
        expected = "*** Found logs in s3:\n***   * s3://bucket/remote/log/location/1.log\nLog line"
        assert actual == expected
        assert metadata == [{"end_of_log": True, "log_pos": 8}]
    def test_read_when_s3_log_missing(self):
        ti = copy.copy(self.ti)
        ti.state = TaskInstanceState.SUCCESS
        log, metadata = self.s3_task_handler.read(ti)
        assert 1 == len(log)
        assert len(log) == len(metadata)
        actual = log[0][0][-1]
        expected = "*** No logs found on s3 for ti=<TaskInstance: dag_for_testing_s3_task_handler.task_for_testing_s3_log_handler test [success]>\n"  
        assert actual == expected
        assert {"end_of_log": True, "log_pos": 0} == metadata[0]
    def test_s3_read_when_log_missing(self):