fix serialize composite block test

Author: limotova

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/data/BlockSerializationTests.java

@@ -8,6 +8,8 @@
 package org.elasticsearch.compute.data;
 
 import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.TransportVersion;
+import org.elasticsearch.TransportVersions;
 import org.elasticsearch.common.breaker.CircuitBreaker;
 import org.elasticsearch.common.breaker.CircuitBreakingException;
 import org.elasticsearch.common.unit.ByteSizeValue;
@@ -22,6 +24,7 @@
 import org.elasticsearch.core.Releasables;
 import org.elasticsearch.indices.breaker.NoneCircuitBreakerService;
 import org.elasticsearch.test.EqualsHashCodeTestUtils;
+import org.elasticsearch.test.TransportVersionUtils;
 
 import java.io.IOException;
 import java.util.Arrays;
@@ -383,7 +386,16 @@ public void testCompositeBlock() throws Exception {
             for (int b = 0; b < numBlocks; b++) {
                 assertThat(origBlock.getBlock(b), equalTo(blocks[b]));
             }
-            try (CompositeBlock deserBlock = serializeDeserializeBlock(origBlock)) {
+            try (
+                CompositeBlock deserBlock = serializeDeserializeBlockWithVersion(
+                    origBlock,
+                    TransportVersionUtils.randomVersionBetween(
+                        random(),
+                        TransportVersions.AGGREGATE_METRIC_DOUBLE_BLOCK,
+                        TransportVersion.current()
+                    )
+                )
+            ) {
                 assertThat(deserBlock.getBlockCount(), equalTo(numBlocks));
                 for (int b = 0; b < numBlocks; b++) {
                     assertThat(deserBlock.getBlock(b), equalTo(origBlock.getBlock(b)));

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/data/SerializationTestCase.java

@@ -53,10 +53,14 @@ BlockStreamInput blockStreamInput(BytesStreamOutput out) {
         return new BlockStreamInput(ByteBufferStreamInput.wrap(BytesReference.toBytes(out.bytes())), blockFactory);
     }
 
-    @SuppressWarnings("unchecked")
     <T extends Block> T serializeDeserializeBlock(T origBlock) throws IOException {
+        TransportVersion version = TransportVersionUtils.randomCompatibleVersion(random());
+        return serializeDeserializeBlockWithVersion(origBlock, version);
+    }
+
+    @SuppressWarnings("unchecked")
+    <T extends Block> T serializeDeserializeBlockWithVersion(T origBlock, TransportVersion version) throws IOException {
         try (BytesStreamOutput out = new BytesStreamOutput()) {
-            TransportVersion version = TransportVersionUtils.randomCompatibleVersion(random());
             out.setTransportVersion(version);
             Block.writeTypedBlock(origBlock, out);
             try (BlockStreamInput in = blockStreamInput(out)) {