[Transform] add debug logging to investigate #62951 (#62990)

x-pack/plugin/transform/qa/single-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/transform/integration/continuous/ContinuousTestCase.java

@@ -86,7 +86,8 @@ public abstract class ContinuousTestCase extends ESRestTestCase {
 
     protected AggregatorFactories.Builder addCommonAggregations(AggregatorFactories.Builder builder) {
         builder.addAggregator(AggregationBuilders.max(MAX_RUN_FIELD).field("run"))
-            .addAggregator(AggregationBuilders.count("count").field("run"));
+            .addAggregator(AggregationBuilders.count("count").field("run"))
+            .addAggregator(AggregationBuilders.max("time.max").field("timestamp"));
         return builder;
     }
 

x-pack/plugin/transform/qa/single-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/transform/integration/continuous/TermsGroupByIT.java

@@ -141,13 +141,19 @@ public class TermsGroupByIT extends ContinuousTestCase {
             );
 
             // test optimization, transform should only rewrite documents that require it
+            // run.ingest is set by the pipeline, run.max is set by the transform
+            // run.ingest > run.max means, the data point has been re-created/re-fed although it wasn't necessary,
+            // this is probably a bug in transform's change collection optimization
+            // run.ingest < run.max means the ingest pipeline wasn't updated, this might be a bug in put pipeline
             assertThat(
                 "Ingest run: "
                     + XContentMapValues.extractValue(INGEST_RUN_FIELD, source)
                     + " did not match max run: "
                     + XContentMapValues.extractValue(MAX_RUN_FIELD, source)
                     + ", iteration: "
-                    + iteration,
+                    + iteration
+                    + " full source: "
+                    + source,
                 // TODO: aggs return double for MAX_RUN_FIELD, although it is an integer
                 Double.valueOf((Integer) XContentMapValues.extractValue(INGEST_RUN_FIELD, source)),
                 equalTo(XContentMapValues.extractValue(MAX_RUN_FIELD, source))

x-pack/plugin/transform/qa/single-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/transform/integration/continuous/TransformContinuousIT.java

@@ -388,6 +388,11 @@ public class TransformContinuousIT extends ESRestTestCase {
     }
 
     private void waitUntilTransformsReachedUpperBound(long timeStampUpperBoundMillis) throws Exception {
+        logger.info(
+            "wait until transform reaches timestamp_millis: {}",
+            ContinuousTestCase.STRICT_DATE_OPTIONAL_TIME_PRINTER_NANOS.withZone(ZoneId.of("UTC"))
+                .format(Instant.ofEpochMilli(timeStampUpperBoundMillis))
+        );
         for (ContinuousTestCase testCase : transformTestCases) {
             assertBusy(() -> {
                 TransformStats stats = getTransformStats(testCase.getName()).getTransformsStats().get(0);

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/TransformIndexer.java

@@ -848,7 +848,7 @@ public abstract class TransformIndexer extends AsyncTwoPhaseIndexer<TransformInd
         // TODO: if buildChangesQuery changes the query it get overwritten
         sourceBuilder.query(filteredQuery);
 
-        logger.trace("running changes query {}", sourceBuilder);
+        logger.debug("[{}] Querying for changes: {}", getJobId(), sourceBuilder);
         return sourceBuilder;
     }