fix date-processor to a new default year for every new pipeline execution. (#22601)

Beforehand, the DateProcessor constructs its joda pattern formatter during processor
construction. This led to newly ingested documents being defaulted to
the year that the pipeline was constructed, not that of processing.

Fixes #22547.

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

@@ -22,6 +22,7 @@ package org.elasticsearch.ingest.common;
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
 import org.joda.time.format.DateTimeFormat;
+import org.joda.time.format.DateTimeFormatter;
 import org.joda.time.format.ISODateTimeFormat;
 
 import java.util.Locale;
@@ -65,9 +66,8 @@ enum DateFormat {
     Joda {
         @Override
         Function<String, DateTime> getFunction(String format, DateTimeZone timezone, Locale locale) {
-            return DateTimeFormat.forPattern(format)
-                .withDefaultYear((new DateTime(DateTimeZone.UTC)).getYear())
-                .withZone(timezone).withLocale(locale)::parseDateTime;
+            DateTimeFormatter parser = DateTimeFormat.forPattern(format).withZone(timezone).withLocale(locale);
+            return text -> parser.withDefaultYear((new DateTime(DateTimeZone.UTC)).getYear()).parseDateTime(text);
         }
     };
 

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

@@ -106,13 +106,13 @@ public class DateProcessorTests extends ESTestCase {
 
     public void testJodaPatternDefaultYear() {
         DateProcessor dateProcessor = new DateProcessor(randomAsciiOfLength(10), DateTimeZone.forID("Europe/Amsterdam"), Locale.ENGLISH,
-                "date_as_string", Collections.singletonList("dd/MM"), "date_as_date");
+            "date_as_string", Collections.singletonList("dd/MM"), "date_as_date");
         Map<String, Object> document = new HashMap<>();
         document.put("date_as_string", "12/06");
         IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), document);
         dateProcessor.execute(ingestDocument);
-        assertThat(ingestDocument.getFieldValue("date_as_date", String.class), equalTo(DateTime.now().getYear() +
-                "-06-12T00:00:00.000+02:00"));
+        assertThat(ingestDocument.getFieldValue("date_as_date", String.class),
+            equalTo(DateTime.now().getYear() + "-06-12T00:00:00.000+02:00"));
     }
 
     public void testTAI64N() {