
**Apache Airflow version**: 2.*

**Apache Airflow Provider versions** (please include all providers that are relevant to your bug): providers-elasticsearch==1.*/2.*



**Kubernetes version (if you are using kubernetes)** (use `kubectl version`): 1.17

**What happened**:
When using the ElasticsearchTaskHandler I see the broken order of the logs in the interface. 
![IMAGE 2021-08-09 18:05:13](https://user-images.githubusercontent.com/3032319/128728656-86736b50-4bd4-4178-9066-b70197b60421.jpg)

raw log entries:
{"asctime": "2021-07-16 09:37:22,278", "filename": "pod_launcher_deprecated.py", "lineno": 131, "levelname": "WARNING", "message": "Pod not yet started: a3ca15e631a7f67697e10b23bae82644.6314229ca0c84d7ba29c870edc39a268", "exc_text": null, "dag_id": "dag_name_AS10_A_A_A_A", "task_id": "k8spod_make_upload_dir_1_2", "execution_date": "2021_07_16T09_10_00_000000", "try_number": "1", "log_id": "dag_name_AS10_A_A_A_A-k8spod_make_upload_dir_1_2-2021_07_16T09_10_00_000000-1", "offset": 1626427744958776832}

{"asctime": "2021-07-16 09:37:26,485", "filename": "taskinstance.py", "lineno": 1191, "levelname": "INFO", "message": "Marking task as SUCCESS. dag_id=pod_gprs_raw_from_nfs_AS10_A_A_A_A, task_id=k8spod_make_upload_dir_1_2, execution_date=20210716T091000, start_date=20210716T092905, end_date=20210716T093726", "exc_text": null, "dag_id": "dag_name_AS10_A_A_A_A", "task_id": "k8spod_make_upload_dir_1_2", "execution_date": "2021_07_16T09_10_00_000000", "try_number": "1", "log_id": "dag_name_AS10_A_A_A_A-k8spod_make_upload_dir_1_2-2021_07_16T09_10_00_000000-1", "offset": 1626427744958776832}


**What you expected to happen**:

The problem lies in the method [set_context](https://github.com/apache/airflow/blob/662cb8c6ac8becb26ff405f8b21acfccdd8de2ae/airflow/providers/elasticsearch/log/es_task_handler.py#L271) that is set on the instance of the class and then all entries in the log go with the same offset, which is used to select for display in the interface. When we redefined method emit and put the offset to the record, the problem disappeared


**How to reproduce it**:
Run a long-lived task that generates logs, in our case it is a spark task launched from a docker container

airflow/providers/elasticsearch/log/es_task_handler.py
import elasticsearch
import pendulum
from elasticsearch_dsl import Search
from airflow.configuration import conf
from airflow.models import TaskInstance
from airflow.utils import timezone
from airflow.utils.log.file_task_handler import FileTaskHandler
from airflow.utils.log.json_formatter import JSONFormatter
from airflow.utils.log.logging_mixin import ExternalLoggingMixin, LoggingMixin
EsLogMsgType = List[Tuple[str, str]]
class ElasticsearchTaskHandler(FileTaskHandler, ExternalLoggingMixin, LoggingMixin):
    """
    ElasticsearchTaskHandler is a python log handler that
    reads logs from Elasticsearch. Note logs are not directly
    indexed into Elasticsearch. Instead, it flushes logs
    into local files. Additional software setup is required
    to index the log into Elasticsearch, such as using
    Filebeat and Logstash.
    To efficiently query and sort Elasticsearch results, we assume each
    log message has a field `log_id` consists of ti primary keys:
    `log_id = {dag_id}-{task_id}-{execution_date}-{try_number}`
    Log messages with specific log_id are sorted based on `offset`,
    which is a unique integer indicates log message's order.
    Timestamp here are unreliable because multiple log messages
    might have the same timestamp.
    """
    PAGE = 0
    MAX_LINE_PER_PAGE = 1000
    LOG_NAME = 'Elasticsearch'
    def __init__(