Fix ingest timezone logic backport(#51215) (#51802)

when a timezone is not provided Ingest logic should consider a time to be in a timezone provided as a parameter.
When a timezone is provided Ingest should recalculate a time to the timezone provided as a parameter

closes #51108
backport(#51215)

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/DateFormat.java

@@ -45,7 +45,9 @@ enum DateFormat {
     Iso8601 {
         @Override
         Function<String, ZonedDateTime> getFunction(String format, ZoneId timezone, Locale locale) {
-            return (date) -> DateFormatters.from(DateFormatter.forPattern("iso8601").parse(date)).withZoneSameInstant(timezone);
+            return (date) -> DateFormatters.from(DateFormatter.forPattern("iso8601").parse(date), timezone)
+                                            .withZoneSameInstant(timezone);
+
         }
     },
     Unix {

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/DateFormatTests.java

@@ -85,14 +85,19 @@ public class DateFormatTests extends ESTestCase {
     }
 
     public void testParseISO8601() {
-        assertThat(DateFormat.Iso8601.getFunction(null, ZoneOffset.UTC, null).apply("2001-01-01T00:00:00-0800").toInstant().toEpochMilli(),
-                equalTo(978336000000L));
-        assertThat(DateFormat.Iso8601.getFunction(null, ZoneOffset.UTC, null).apply("2001-01-01T00:00:00-0800").toString(),
-                equalTo("2001-01-01T08:00Z"));
+        assertThat(DateFormat.Iso8601.getFunction(null, ZoneOffset.UTC, null).apply("2001-01-01T00:00:00-0800")
+                                     .toInstant().toEpochMilli(), equalTo(978336000000L));
         assertThat(DateFormat.Iso8601.getFunction(null, ZoneOffset.UTC, null).apply("2001-01-01T00:00:00-0800").toString(),
                 equalTo("2001-01-01T08:00Z"));
     }
 
+    public void testParseWhenZoneNotPresentInText() {
+        assertThat(DateFormat.Iso8601.getFunction(null, ZoneOffset.of("+0100"), null).apply("2001-01-01T00:00:00")
+                                     .toInstant().toEpochMilli(), equalTo(978303600000L));
+        assertThat(DateFormat.Iso8601.getFunction(null, ZoneOffset.of("+0100"), null).apply("2001-01-01T00:00:00").toString(),
+            equalTo("2001-01-01T00:00+01:00"));
+    }
+
     public void testParseISO8601Failure() {
         Function<String, ZonedDateTime> function = DateFormat.Iso8601.getFunction(null, ZoneOffset.UTC, null);
         try {

modules/ingest-common/src/test/resources/rest-api-spec/test/ingest/30_date_processor.yml

@@ -40,6 +40,59 @@ teardown:
   - match: { _source.date_source_field: "12/06/2010" }
   - match: { _source.date_target_field: "2010-06-12T00:00:00.000+02:00" }
 
+
+---
+"Test date processor timezone calculations":
+  - do:
+      ingest.put_pipeline:
+        id: "my_pipeline_2"
+        body:  >
+          {
+            "description": "_description",
+            "processors": [
+              {
+                "date" : {
+                  "field" : "date_source_field",
+                  "target_field" : "date_target_field",
+                  "formats" : ["ISO8601"],
+                  "timezone" : "+01:00"
+                }
+              }
+            ]
+          }
+  - match: { acknowledged: true }
+
+
+  - do:
+      index:
+        index: test2
+        id: 1
+        pipeline: "my_pipeline_2"
+        body: {date_source_field: "2010-06-01T00:00:00.000"}
+
+  - do:
+      get:
+        index: test2
+        id: 1
+  - match: { _source.date_source_field: "2010-06-01T00:00:00.000" }
+  # date field without a timezone gets timezone from a pipeline
+  - match: { _source.date_target_field: "2010-06-01T00:00:00.000+01:00" }
+
+  - do:
+      index:
+        index: test2
+        id: 2
+        pipeline: "my_pipeline_2"
+        body: {date_source_field: "2010-06-01T00:00:00.000Z"}
+
+  - do:
+      get:
+        index: test2
+        id: 2
+  - match: { _source.date_source_field: "2010-06-01T00:00:00.000Z" }
+   # date field with a timezone has its time recalculated to a target timezone from a pipeline
+  - match: { _source.date_target_field: "2010-06-01T01:00:00.000+01:00" }
+
 ---
 "Test date processor with no timezone configured":
 

server/src/main/java/org/elasticsearch/common/time/DateFormatters.java

@@ -1836,13 +1836,17 @@ public class DateFormatters {
      * @return The converted zoned date time
      */
     public static ZonedDateTime from(TemporalAccessor accessor) {
+        return from(accessor, ZoneOffset.UTC);
+    }
+
+    public static ZonedDateTime from(TemporalAccessor accessor, ZoneId defaultZone) {
         if (accessor instanceof ZonedDateTime) {
             return (ZonedDateTime) accessor;
         }
 
         ZoneId zoneId = accessor.query(TemporalQueries.zone());
         if (zoneId == null) {
-            zoneId = ZoneOffset.UTC;
+            zoneId = defaultZone;
         }
 
         LocalDate localDate = accessor.query(LOCAL_DATE_QUERY);