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

self.conn.put_object(Bucket="bucket", Key=self.remote_log_key, Body=b"previous ")
        self.s3_task_handler.s3_write("text", self.remote_log_location)
        body = boto3.resource("s3").Object("bucket", self.remote_log_key).get()["Body"].read()
        assert body == b"previous \ntext"
    def test_write_raises(self):
        handler = self.s3_task_handler
        url = "s3://nonexistentbucket/foo"
        with mock.patch.object(handler.log, "error") as mock_error:
            handler.s3_write("text", url)
            mock_error.assert_called_once_with("Could not write logs to %s", url, exc_info=True)
    def test_close(self):
        self.s3_task_handler.set_context(self.ti)
        assert self.s3_task_handler.upload_on_close
        self.s3_task_handler.close()
        
        boto3.resource("s3").Object("bucket", self.remote_log_key).get()
    def test_close_no_upload(self):
        self.ti.raw = True
        self.s3_task_handler.set_context(self.ti)
        assert not self.s3_task_handler.upload_on_close
        self.s3_task_handler.close()
        with pytest.raises(ClientError):
            boto3.resource("s3").Object("bucket", self.remote_log_key).get()