Merge branch 'main' into 2025/02/05/validate-known-transport-handshake-version

Author: DaveCTurner

muted-tests.yml

@@ -134,6 +134,12 @@ tests:
 - class: org.elasticsearch.datastreams.DataStreamsClientYamlTestSuiteIT
   method: test {p0=data_stream/120_data_streams_stats/Multiple data stream}
   issue: https://github.com/elastic/elasticsearch/issues/118217
+  # TODO: re-enable after backporting https://github.com/elastic/elasticsearch/pull/119110
+- class: org.elasticsearch.test.rest.ClientYamlTestSuiteIT
+  method: test {yaml=update/100_synthetic_source/keyword}
+  # TODO: re-enable after backporting https://github.com/elastic/elasticsearch/pull/119110
+- class: org.elasticsearch.test.rest.ClientYamlTestSuiteIT
+  method: test {yaml=update/100_synthetic_source/stored text}
 - class: org.elasticsearch.xpack.searchablesnapshots.RetrySearchIntegTests
   method: testSearcherId
   issue: https://github.com/elastic/elasticsearch/issues/118374

qa/smoke-test-multinode/build.gradle

@@ -28,5 +28,7 @@ tasks.named("yamlRestTest").configure {
     'cat.templates/10_basic/No templates',
     'cat.templates/10_basic/Sort templates',
     'cat.templates/10_basic/Multiple template',
+    'update/100_synthetic_source/keyword',
+    'update/100_synthetic_source/stored text'
   ].join(',')
 }

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/update/100_synthetic_source.yml

@@ -6,8 +6,8 @@ setup:
 ---
 keyword:
   - requires:
-      cluster_features: ["gte_v8.4.0"]
-      reason: introduced in 8.4.0
+      cluster_features: [ "mapper.synthetic_recovery_source" ]
+      reason: requires synthetic recovery source
 
   - do:
       indices.create:
@@ -60,13 +60,14 @@ keyword:
         index: test
         run_expensive_tasks: true
   - is_false: test.fields._source
-  - is_true: test.fields._recovery_source
+  # When synthetic source is used there is no _recovery_source field
+  - match: { test.fields._recovery_source: null }
 
 ---
 stored text:
   - requires:
-      cluster_features: ["gte_v8.5.0"]
-      reason: introduced in 8.5.0
+      cluster_features: [ "mapper.synthetic_recovery_source" ]
+      reason: requires synthetic recovery source
 
   - do:
       indices.create:
@@ -121,4 +122,5 @@ stored text:
         index: test
         run_expensive_tasks: true
   - is_false: test.fields._source
-  - is_true: test.fields._recovery_source
+  # When synthetic source is used there is no _recovery_source field
+  - match: { test.fields._recovery_source: null }

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -175,6 +175,8 @@ static TransportVersion def(int id) {
     public static final TransportVersion COHERE_BIT_EMBEDDING_TYPE_SUPPORT_ADDED_BACKPORT_8_X = def(8_840_0_01);
     public static final TransportVersion ELASTICSEARCH_9_0 = def(9_000_0_00);
     public static final TransportVersion COHERE_BIT_EMBEDDING_TYPE_SUPPORT_ADDED = def(9_001_0_00);
+    public static final TransportVersion TRANSPORT_STATS_HANDLING_TIME_REQUIRED = def(9_002_0_00);
+
     /*
      * STOP! READ THIS FIRST! No, really,
      *        ____ _____ ___  ____  _        ____  _____    _    ____    _____ _   _ ___ ____    _____ ___ ____  ____ _____ _

server/src/main/java/org/elasticsearch/common/settings/Setting.java

@@ -1580,6 +1580,15 @@ public static Setting<Boolean> boolSetting(String key, boolean defaultValue, Val
         return new Setting<>(key, Boolean.toString(defaultValue), booleanParser(key, properties), validator, properties);
     }
 
+    public static Setting<Boolean> boolSetting(
+        String key,
+        Function<Settings, String> defaultValueFn,
+        Validator<Boolean> validator,
+        Property... properties
+    ) {
+        return new Setting<>(key, defaultValueFn, booleanParser(key, properties), validator, properties);
+    }
+
     public static Setting<Boolean> boolSetting(String key, Function<Settings, String> defaultValueFn, Property... properties) {
         return new Setting<>(key, defaultValueFn, booleanParser(key, properties), properties);
     }

server/src/main/java/org/elasticsearch/index/IndexSettings.java

@@ -24,6 +24,7 @@
 import org.elasticsearch.common.time.DateUtils;
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.common.util.FeatureFlag;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.mapper.IgnoredSourceFieldMapper;
 import org.elasticsearch.index.mapper.Mapper;
@@ -39,6 +40,7 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 
@@ -722,9 +724,25 @@ public Iterator<Setting<?>> settings() {
         Setting.Property.ServerlessPublic
     );
 
+    public static final FeatureFlag RECOVERY_USE_SYNTHETIC_SOURCE = new FeatureFlag("index_recovery_use_synthetic_source");
     public static final Setting<Boolean> RECOVERY_USE_SYNTHETIC_SOURCE_SETTING = Setting.boolSetting(
         "index.recovery.use_synthetic_source",
-        false,
+        settings -> {
+            boolean isSyntheticSourceRecoveryFeatureFlagEnabled = RECOVERY_USE_SYNTHETIC_SOURCE.isEnabled();
+            boolean isNewIndexVersion = SETTING_INDEX_VERSION_CREATED.get(settings)
+                .onOrAfter(IndexVersions.USE_SYNTHETIC_SOURCE_FOR_RECOVERY_BY_DEFAULT);
+            boolean isIndexVersionInBackportRange = SETTING_INDEX_VERSION_CREATED.get(settings)
+                .between(IndexVersions.USE_SYNTHETIC_SOURCE_FOR_RECOVERY_BY_DEFAULT_BACKPORT, IndexVersions.UPGRADE_TO_LUCENE_10_0_0);
+
+            boolean useSyntheticRecoverySource = isSyntheticSourceRecoveryFeatureFlagEnabled
+                && (isNewIndexVersion || isIndexVersionInBackportRange);
+
+            return String.valueOf(
+                useSyntheticRecoverySource
+                    && Objects.equals(INDEX_MAPPER_SOURCE_MODE_SETTING.get(settings), SourceFieldMapper.Mode.SYNTHETIC)
+            );
+
+        },
         new Setting.Validator<>() {
             @Override
             public void validate(Boolean value) {}
@@ -1083,7 +1101,8 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
         skipIgnoredSourceRead = scopedSettings.get(IgnoredSourceFieldMapper.SKIP_IGNORED_SOURCE_READ_SETTING);
         indexMappingSourceMode = scopedSettings.get(INDEX_MAPPER_SOURCE_MODE_SETTING);
         recoverySourceEnabled = RecoverySettings.INDICES_RECOVERY_SOURCE_ENABLED_SETTING.get(nodeSettings);
-        recoverySourceSyntheticEnabled = scopedSettings.get(RECOVERY_USE_SYNTHETIC_SOURCE_SETTING);
+        recoverySourceSyntheticEnabled = DiscoveryNode.isStateless(nodeSettings) == false
+            && scopedSettings.get(RECOVERY_USE_SYNTHETIC_SOURCE_SETTING);
         if (recoverySourceSyntheticEnabled) {
             if (DiscoveryNode.isStateless(settings)) {
                 throw new IllegalArgumentException("synthetic recovery source is only allowed in stateful");

server/src/main/java/org/elasticsearch/index/IndexVersions.java

@@ -134,6 +134,7 @@ private static Version parseUnchecked(String version) {
     public static final IndexVersion UPGRADE_TO_LUCENE_9_12_1 = def(8_523_0_00, parseUnchecked("9.12.1"));
     public static final IndexVersion INFERENCE_METADATA_FIELDS_BACKPORT = def(8_524_0_00, parseUnchecked("9.12.1"));
     public static final IndexVersion LOGSB_OPTIONAL_SORTING_ON_HOST_NAME_BACKPORT = def(8_525_0_00, parseUnchecked("9.12.1"));
+    public static final IndexVersion USE_SYNTHETIC_SOURCE_FOR_RECOVERY_BY_DEFAULT_BACKPORT = def(8_526_0_00, parseUnchecked("9.12.1"));
     public static final IndexVersion UPGRADE_TO_LUCENE_10_0_0 = def(9_000_0_00, Version.LUCENE_10_0_0);
     public static final IndexVersion LOGSDB_DEFAULT_IGNORE_DYNAMIC_BEYOND_LIMIT = def(9_001_0_00, Version.LUCENE_10_0_0);
     public static final IndexVersion TIME_BASED_K_ORDERED_DOC_ID = def(9_002_0_00, Version.LUCENE_10_0_0);
@@ -144,6 +145,7 @@ private static Version parseUnchecked(String version) {
     public static final IndexVersion SOURCE_MAPPER_MODE_ATTRIBUTE_NOOP = def(9_007_0_00, Version.LUCENE_10_0_0);
     public static final IndexVersion HOSTNAME_DOC_VALUES_SPARSE_INDEX = def(9_008_0_00, Version.LUCENE_10_0_0);
     public static final IndexVersion UPGRADE_TO_LUCENE_10_1_0 = def(9_009_0_00, Version.LUCENE_10_1_0);
+    public static final IndexVersion USE_SYNTHETIC_SOURCE_FOR_RECOVERY_BY_DEFAULT = def(9_010_00_0, Version.LUCENE_10_1_0);
     /*
      * STOP! READ THIS FIRST! No, really,
      *        ____ _____ ___  ____  _        ____  _____    _    ____    _____ _   _ ___ ____    _____ ___ ____  ____ _____ _

server/src/main/java/org/elasticsearch/transport/TransportStats.java

@@ -18,7 +18,6 @@
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.xcontent.ChunkedToXContent;
 import org.elasticsearch.core.TimeValue;
-import org.elasticsearch.core.UpdateForV9;
 import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentBuilder;
 
@@ -70,18 +69,16 @@ public TransportStats(StreamInput in) throws IOException {
         rxSize = in.readVLong();
         txCount = in.readVLong();
         txSize = in.readVLong();
-        if (in.getTransportVersion().onOrAfter(TransportVersions.V_8_1_0) && in.readBoolean()) {
-            inboundHandlingTimeBucketFrequencies = new long[HandlingTimeTracker.BUCKET_COUNT];
-            for (int i = 0; i < inboundHandlingTimeBucketFrequencies.length; i++) {
-                inboundHandlingTimeBucketFrequencies[i] = in.readVLong();
-            }
-            outboundHandlingTimeBucketFrequencies = new long[HandlingTimeTracker.BUCKET_COUNT];
-            for (int i = 0; i < inboundHandlingTimeBucketFrequencies.length; i++) {
-                outboundHandlingTimeBucketFrequencies[i] = in.readVLong();
-            }
-        } else {
-            inboundHandlingTimeBucketFrequencies = new long[0];
-            outboundHandlingTimeBucketFrequencies = new long[0];
+        if (in.getTransportVersion().before(TransportVersions.TRANSPORT_STATS_HANDLING_TIME_REQUIRED)) {
+            in.readBoolean();
+        }
+        inboundHandlingTimeBucketFrequencies = new long[HandlingTimeTracker.BUCKET_COUNT];
+        for (int i = 0; i < inboundHandlingTimeBucketFrequencies.length; i++) {
+            inboundHandlingTimeBucketFrequencies[i] = in.readVLong();
+        }
+        outboundHandlingTimeBucketFrequencies = new long[HandlingTimeTracker.BUCKET_COUNT];
+        for (int i = 0; i < inboundHandlingTimeBucketFrequencies.length; i++) {
+            outboundHandlingTimeBucketFrequencies[i] = in.readVLong();
         }
         if (in.getTransportVersion().onOrAfter(TransportVersions.V_8_8_0)) {
             transportActionStats = Collections.unmodifiableMap(in.readOrderedMap(StreamInput::readString, TransportActionStats::new));
@@ -99,15 +96,16 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeVLong(rxSize);
         out.writeVLong(txCount);
         out.writeVLong(txSize);
-        if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_1_0)) {
-            assert (inboundHandlingTimeBucketFrequencies.length > 0) == (outboundHandlingTimeBucketFrequencies.length > 0);
-            out.writeBoolean(inboundHandlingTimeBucketFrequencies.length > 0);
-            for (long handlingTimeBucketFrequency : inboundHandlingTimeBucketFrequencies) {
-                out.writeVLong(handlingTimeBucketFrequency);
-            }
-            for (long handlingTimeBucketFrequency : outboundHandlingTimeBucketFrequencies) {
-                out.writeVLong(handlingTimeBucketFrequency);
-            }
+        assert inboundHandlingTimeBucketFrequencies.length == HandlingTimeTracker.BUCKET_COUNT;
+        assert outboundHandlingTimeBucketFrequencies.length == HandlingTimeTracker.BUCKET_COUNT;
+        if (out.getTransportVersion().before(TransportVersions.TRANSPORT_STATS_HANDLING_TIME_REQUIRED)) {
+            out.writeBoolean(true);
+        }
+        for (long handlingTimeBucketFrequency : inboundHandlingTimeBucketFrequencies) {
+            out.writeVLong(handlingTimeBucketFrequency);
+        }
+        for (long handlingTimeBucketFrequency : outboundHandlingTimeBucketFrequencies) {
+            out.writeVLong(handlingTimeBucketFrequency);
         }
         if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_8_0)) {
             out.writeMap(transportActionStats, StreamOutput::writeWriteable);
@@ -166,24 +164,13 @@ public Map<String, TransportActionStats> getTransportActionStats() {
         return transportActionStats;
     }
 
-    @UpdateForV9(owner = UpdateForV9.Owner.DISTRIBUTED_COORDINATION)
-    // Review and simplify the if-else blocks containing this symbol once v9 is released
-    private static final boolean IMPOSSIBLE_IN_V9 = true;
-
     private boolean assertHistogramsConsistent() {
         assert inboundHandlingTimeBucketFrequencies.length == outboundHandlingTimeBucketFrequencies.length;
-        if (inboundHandlingTimeBucketFrequencies.length == 0) {
-            // Stats came from before v8.1
-            assert IMPOSSIBLE_IN_V9;
-        } else {
-            assert inboundHandlingTimeBucketFrequencies.length == HandlingTimeTracker.BUCKET_COUNT;
-        }
+        assert inboundHandlingTimeBucketFrequencies.length == HandlingTimeTracker.BUCKET_COUNT;
         return true;
     }
 
     @Override
-    @UpdateForV9(owner = UpdateForV9.Owner.DISTRIBUTED_COORDINATION)
-    // review the "if" blocks checking for non-empty once we have
     public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params outerParams) {
         return Iterators.concat(Iterators.single((builder, params) -> {
             builder.startObject(Fields.TRANSPORT);
@@ -193,19 +180,10 @@ public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params outerP
             builder.humanReadableField(Fields.RX_SIZE_IN_BYTES, Fields.RX_SIZE, ByteSizeValue.ofBytes(rxSize));
             builder.field(Fields.TX_COUNT, txCount);
             builder.humanReadableField(Fields.TX_SIZE_IN_BYTES, Fields.TX_SIZE, ByteSizeValue.ofBytes(txSize));
-            if (inboundHandlingTimeBucketFrequencies.length > 0) {
-                histogramToXContent(builder, inboundHandlingTimeBucketFrequencies, Fields.INBOUND_HANDLING_TIME_HISTOGRAM);
-                histogramToXContent(builder, outboundHandlingTimeBucketFrequencies, Fields.OUTBOUND_HANDLING_TIME_HISTOGRAM);
-            } else {
-                // Stats came from before v8.1
-                assert IMPOSSIBLE_IN_V9;
-            }
-            if (transportActionStats.isEmpty() == false) {
-                builder.startObject(Fields.ACTIONS);
-            } else {
-                // Stats came from before v8.8
-                assert IMPOSSIBLE_IN_V9;
-            }
+            assert inboundHandlingTimeBucketFrequencies.length > 0;
+            histogramToXContent(builder, inboundHandlingTimeBucketFrequencies, Fields.INBOUND_HANDLING_TIME_HISTOGRAM);
+            histogramToXContent(builder, outboundHandlingTimeBucketFrequencies, Fields.OUTBOUND_HANDLING_TIME_HISTOGRAM);
+            builder.startObject(Fields.ACTIONS);
             return builder;
         }),
 
@@ -215,12 +193,7 @@ public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params outerP
                 return builder;
             }),
 
-            Iterators.single((builder, params) -> {
-                if (transportActionStats.isEmpty() == false) {
-                    builder.endObject();
-                }
-                return builder.endObject();
-            })
+            Iterators.single((builder, params) -> { return builder.endObject().endObject(); })
         );
     }
 

server/src/test/java/org/elasticsearch/index/mapper/IgnoredSourceFieldMapperTests.java

@@ -2427,8 +2427,14 @@ protected void validateRoundTripReader(String syntheticSource, DirectoryReader r
         // and since the copy is exact, contents of ignored source are different.
         assertReaderEquals(
             "round trip " + syntheticSource,
-            new FieldMaskingReader(Set.of(SourceFieldMapper.RECOVERY_SOURCE_NAME, IgnoredSourceFieldMapper.NAME), reader),
-            new FieldMaskingReader(Set.of(SourceFieldMapper.RECOVERY_SOURCE_NAME, IgnoredSourceFieldMapper.NAME), roundTripReader)
+            new FieldMaskingReader(
+                Set.of(SourceFieldMapper.RECOVERY_SOURCE_NAME, IgnoredSourceFieldMapper.NAME, SourceFieldMapper.RECOVERY_SOURCE_SIZE_NAME),
+                reader
+            ),
+            new FieldMaskingReader(
+                Set.of(SourceFieldMapper.RECOVERY_SOURCE_NAME, IgnoredSourceFieldMapper.NAME, SourceFieldMapper.RECOVERY_SOURCE_SIZE_NAME),
+                roundTripReader
+            )
         );
     }
 }