Rollup ignores time_zone on date histogram (elastic#40844)

When translating the original aggregation for the rollup indices,
the timezone of the date histogram is validated against the rollup
job but the value is not copied in the newly created date_histogram.

Author: jimczi

x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/RollupRequestTranslator.java

@@ -224,6 +224,9 @@ private static List<AggregationBuilder> translateDateHistogram(DateHistogramAggr
             filterConditions.add(new TermQueryBuilder(RollupField.formatFieldName(source,
                 DateHistogramGroupConfig.TIME_ZONE), timezone));
 
+            if (source.timeZone() != null) {
+                rolledDateHisto.timeZone(source.timeZone());
+            }
             rolledDateHisto.offset(source.offset());
             if (source.extendedBounds() != null) {
                 rolledDateHisto.extendedBounds(source.extendedBounds());

x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/RollupRequestTranslationTests.java

@@ -31,6 +31,7 @@
 import org.junit.Before;
 
 import java.io.IOException;
+import java.time.ZoneId;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -271,6 +272,37 @@ public void testDateHistoLongIntervalWithMinMax() {
         }
     }
 
+    public void testDateHistoWithTimezone() {
+        ZoneId timeZone = ZoneId.of(randomFrom(ZoneId.getAvailableZoneIds()));
+        DateHistogramAggregationBuilder histo = new DateHistogramAggregationBuilder("test_histo");
+        histo.interval(86400000)
+            .field("foo")
+            .timeZone(timeZone);
+        List<QueryBuilder> filterConditions = new ArrayList<>();
+
+        List<AggregationBuilder> translated = translateAggregation(histo, filterConditions, namedWriteableRegistry);
+        assertThat(translated.size(), equalTo(1));
+        assertThat(translated.get(0), instanceOf(DateHistogramAggregationBuilder.class));
+        DateHistogramAggregationBuilder translatedHisto = (DateHistogramAggregationBuilder)translated.get(0);
+
+        assertThat(translatedHisto.interval(), equalTo(86400000L));
+        assertThat(translatedHisto.field(), equalTo("foo.date_histogram.timestamp"));
+        assertThat(translatedHisto.timeZone(), equalTo(timeZone));
+        assertThat(filterConditions.size(), equalTo(1));
+
+        for (QueryBuilder q : filterConditions) {
+            if (q instanceof TermQueryBuilder) {
+                if (((TermQueryBuilder) q).fieldName().equals("foo.date_histogram.time_zone")) {
+                    assertThat(((TermQueryBuilder) q).value(), equalTo(timeZone.toString()));
+                }  else {
+                    fail("Unexpected Term Query in filter conditions: [" + ((TermQueryBuilder) q).fieldName() + "]");
+                }
+            } else {
+                fail("Unexpected query builder in filter conditions");
+            }
+        }
+    }
+
     public void testAvgMetric() {
         List<QueryBuilder> filterConditions = new ArrayList<>();
         List<AggregationBuilder> translated = translateAggregation(new AvgAggregationBuilder("test_metric")