Support serialization of aggregate metric double literal

Author: dnhatn

server/src/main/java/org/elasticsearch/common/io/stream/StreamOutput.java

@@ -50,7 +50,6 @@
 import java.util.function.IntFunction;
 
 import static java.util.Map.entry;
-import static org.elasticsearch.TransportVersions.V_8_11_X;
 
 /**
  * A stream from another node to this node. Technically, it can also be streamed from a byte array but that is mostly for testing.
@@ -813,13 +812,10 @@ public final void writeOptionalInstant(@Nullable Instant instant) throws IOExcep
         entry(GenericNamedWriteable.class, (o, v) -> {
             // Note that we do not rely on the checks in VersionCheckingStreamOutput because that only applies to CCS
             final var genericNamedWriteable = (GenericNamedWriteable) v;
-            TransportVersion minSupportedVersion = genericNamedWriteable.getMinimalSupportedVersion();
-            assert minSupportedVersion.onOrAfter(V_8_11_X) : "[GenericNamedWriteable] requires [" + V_8_11_X + "]";
-            if (o.getTransportVersion().before(minSupportedVersion)) {
+            if (genericNamedWriteable.supportsVersion(o.getTransportVersion()) == false) {
                 final var message = Strings.format(
-                    "[%s] requires minimal transport version [%s] and cannot be sent using transport version [%s]",
+                    "[%s] doesn't support serialization with transport version [%s]",
                     genericNamedWriteable.getWriteableName(),
-                    minSupportedVersion,
                     o.getTransportVersion()
                 );
                 assert false : message;

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/data/AggregateMetricDoubleBlockBuilder.java

@@ -200,9 +200,14 @@ public void writeTo(StreamOutput out) throws IOException {
         }
 
         @Override
-        public TransportVersion getMinimalSupportedVersion() {
-            return TransportVersions.ESQL_AGGREGATE_METRIC_DOUBLE_LITERAL;
+        public boolean supportsVersion(TransportVersion version) {
+            return version.onOrAfter(TransportVersions.ESQL_AGGREGATE_METRIC_DOUBLE_LITERAL);
         }
 
+        @Override
+        public TransportVersion getMinimalSupportedVersion() {
+            assert false : "must not be called when overriding supportsVersion";
+            throw new UnsupportedOperationException("must not be called when overriding supportsVersion");
+        }
     }
 }