fix date-processor to a new default year for every new pipeline execution. (elastic#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 elastic#22547.

Author: talevy

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

@@ -22,6 +22,7 @@
 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 @@ private long parseMillis(String date) {
     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 void testJodaPatternLocale() {
 
     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() {