[AIRFLOW-3370] Add stdout output options to Elasticsearch task log ha…
diff --git a/airflow/utils/log/es_task_handler.py b/airflow/utils/log/es_task_handler.py
index bc2c38aacd615..ae2025ce03708 100644
--- a/airflow/utils/log/es_task_handler.py
+++ b/airflow/utils/log/es_task_handler.py
@@ -53,7 +53,7 @@ class ElasticsearchTaskHandler(FileTaskHandler, LoggingMixin):
 
     def __init__(self, base_log_folder, filename_template,
                  log_id_template, end_of_log_mark,
-                 write_stdout, json_format, record_labels,
+                 write_stdout, json_format, json_fields,
                  host='localhost:9200'):
         """
         :param base_log_folder: base folder to store logs locally
@@ -73,7 +73,7 @@ def __init__(self, base_log_folder, filename_template,
         self.end_of_log_mark = end_of_log_mark
         self.write_stdout = write_stdout
         self.json_format = json_format
-        self.record_labels = [label.strip() for label in record_labels.split(",")]
+        self.json_fields = [label.strip() for label in json_fields.split(",")]
         self.handler = None
 
     def _render_log_id(self, ti, try_number):
diff --git a/tests/utils/log/test_es_task_handler.py b/tests/utils/log/test_es_task_handler.py
index 7b501a2c8e93c..56428ec82548b 100644
--- a/tests/utils/log/test_es_task_handler.py
+++ b/tests/utils/log/test_es_task_handler.py
@@ -230,6 +230,15 @@ def test_set_context(self):
         self.es_task_handler.set_context(self.ti)
         self.assertTrue(self.es_task_handler.mark_end_on_close)
 
+    def test_set_context_w_json_format_and_write_stdout(self):
+        self.es_task_handler.formatter = mock.MagicMock()
+        self.es_task_handler.formatter._fmt = mock.MagicMock()
+        self.es_task_handler.formatter._fmt.find = mock.MagicMock(return_value=1)
+        self.es_task_handler.writer = mock.MagicMock()
+        self.es_task_handler.write_stdout = True
+        self.es_task_handler.json_format = True
+        self.es_task_handler.set_context(self.ti)
+
     def test_close(self):
         self.es_task_handler.set_context(self.ti)
         self.es_task_handler.close()

### Jira

- [x] My PR addresses the following [Airflow Jira](https://issues.apache.org/jira/browse/AIRFLOW/) issues and references them in the PR title. For example, "\[AIRFLOW-3370\] My Airflow PR"
  - https://issues.apache.org/jira/browse/AIRFLOW-3370

### Description

Fix inconsistency in ElasticsearchTaskHandler class.
Rename `record_labels` to `json_fields`


- [x] Here are some details about my PR, including screenshots of any UI changes:

### Code Quality

- [x] Passes `flake8`
