diff --git a/modules/reindex/src/test/java/org/elasticsearch/reindex/RethrottleRequestWireSerializingTests.java b/modules/reindex/src/test/java/org/elasticsearch/reindex/RethrottleRequestWireSerializingTests.java new file mode 100644 index 0000000000000..a37543fb36e42 --- /dev/null +++ b/modules/reindex/src/test/java/org/elasticsearch/reindex/RethrottleRequestWireSerializingTests.java @@ -0,0 +1,140 @@ +/* + * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one + * or more contributor license agreements. Licensed under the "Elastic License + * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side + * Public License v 1"; you may not use this file except in compliance with, at + * your election, the "Elastic License 2.0", the "GNU Affero General Public + * License v3.0 only", or the "Server Side Public License, v 1". + */ + +package org.elasticsearch.reindex; + +import org.elasticsearch.common.io.stream.StreamInput; +import org.elasticsearch.common.io.stream.StreamOutput; +import org.elasticsearch.common.io.stream.Writeable; +import org.elasticsearch.core.TimeValue; +import org.elasticsearch.index.reindex.DeleteByQueryAction; +import org.elasticsearch.index.reindex.ReindexAction; +import org.elasticsearch.index.reindex.UpdateByQueryAction; +import org.elasticsearch.tasks.TaskId; +import org.elasticsearch.test.AbstractWireSerializingTestCase; + +import java.io.IOException; +import java.util.Arrays; +import java.util.Objects; + +public class RethrottleRequestWireSerializingTests extends AbstractWireSerializingTestCase { + + @Override + protected Writeable.Reader instanceReader() { + return Wrapper::new; + } + + @Override + protected Wrapper createTestInstance() { + RethrottleRequest request = new RethrottleRequest(); + request.setRequestsPerSecond((float) randomDoubleBetween(0.001d, Float.POSITIVE_INFINITY, false)); + if (randomBoolean()) { + request.setParentTask(new TaskId(randomAlphaOfLength(6), randomNonNegativeLong())); + } + if (randomBoolean()) { + request.setTargetTaskId(new TaskId(randomAlphaOfLength(6), randomNonNegativeLong())); + } + if (randomBoolean()) { + request.setTargetParentTaskId(new TaskId(randomAlphaOfLength(6), randomNonNegativeLong())); + } + if (randomBoolean()) { + request.setNodes(randomAlphaOfLength(5)); + } + if (randomBoolean()) { + request.setActions(randomFrom(ReindexAction.NAME, UpdateByQueryAction.NAME, DeleteByQueryAction.NAME)); + } + if (randomBoolean()) { + request.setTimeout(TimeValue.timeValueMillis(between(1, 600_000))); + } + return new Wrapper(request); + } + + @Override + protected Wrapper mutateInstance(Wrapper instance) throws IOException { + RethrottleRequest orig = instance.request; + RethrottleRequest copy = copyInstance(instance).request; + switch (between(0, 6)) { + case 0 -> copy.setParentTask( + randomValueOtherThan(orig.getParentTask(), () -> new TaskId(randomAlphaOfLength(9), randomNonNegativeLong())) + ); + case 1 -> copy.setTargetTaskId( + randomValueOtherThan(orig.getTargetTaskId(), () -> new TaskId(randomAlphaOfLength(9), randomNonNegativeLong())) + ); + case 2 -> copy.setTargetParentTaskId( + randomValueOtherThan(orig.getTargetParentTaskId(), () -> new TaskId(randomAlphaOfLength(9), randomNonNegativeLong())) + ); + case 3 -> copy.setNodes( + randomArrayOtherThan(orig.getNodes(), () -> new String[] { randomAlphaOfLength(8), randomAlphaOfLength(8) }) + ); + case 4 -> copy.setActions( + randomArrayOtherThan( + orig.getActions(), + () -> randomFrom( + new String[] { ReindexAction.NAME }, + new String[] { UpdateByQueryAction.NAME }, + new String[] { DeleteByQueryAction.NAME }, + new String[] { ReindexAction.NAME, UpdateByQueryAction.NAME } + ) + ) + ); + case 5 -> copy.setTimeout(randomValueOtherThan(orig.getTimeout(), () -> TimeValue.timeValueMillis(between(1, 900_000)))); + case 6 -> copy.setRequestsPerSecond( + randomValueOtherThan(orig.getRequestsPerSecond(), () -> (float) randomDoubleBetween(0.001d, 1000d, false)) + ); + default -> throw new AssertionError(); + } + return new Wrapper(copy); + } + + static final class Wrapper implements Writeable { + private final RethrottleRequest request; + + Wrapper(RethrottleRequest request) { + this.request = request; + } + + Wrapper(StreamInput in) throws IOException { + this.request = new RethrottleRequest(in); + } + + @Override + public void writeTo(StreamOutput out) throws IOException { + request.writeTo(out); + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + Wrapper wrapper = (Wrapper) o; + RethrottleRequest a = request; + RethrottleRequest b = wrapper.request; + return Objects.equals(a.getParentTask(), b.getParentTask()) + && Objects.equals(a.getTargetTaskId(), b.getTargetTaskId()) + && Objects.equals(a.getTargetParentTaskId(), b.getTargetParentTaskId()) + && Arrays.equals(a.getNodes(), b.getNodes()) + && Arrays.equals(a.getActions(), b.getActions()) + && Objects.equals(a.getTimeout(), b.getTimeout()) + && Float.compare(a.getRequestsPerSecond(), b.getRequestsPerSecond()) == 0; + } + + @Override + public int hashCode() { + return Objects.hash( + request.getParentTask(), + request.getTargetTaskId(), + request.getTargetParentTaskId(), + Arrays.hashCode(request.getNodes()), + Arrays.hashCode(request.getActions()), + request.getTimeout(), + request.getRequestsPerSecond() + ); + } + } +} diff --git a/modules/reindex/src/test/java/org/elasticsearch/reindex/RoundTripTests.java b/modules/reindex/src/test/java/org/elasticsearch/reindex/RoundTripTests.java deleted file mode 100644 index a7685c052715e..0000000000000 --- a/modules/reindex/src/test/java/org/elasticsearch/reindex/RoundTripTests.java +++ /dev/null @@ -1,248 +0,0 @@ -/* - * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one - * or more contributor license agreements. Licensed under the "Elastic License - * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side - * Public License v 1"; you may not use this file except in compliance with, at - * your election, the "Elastic License 2.0", the "GNU Affero General Public - * License v3.0 only", or the "Server Side Public License, v 1". - */ - -package org.elasticsearch.reindex; - -import org.elasticsearch.TransportVersion; -import org.elasticsearch.common.Strings; -import org.elasticsearch.common.bytes.BytesArray; -import org.elasticsearch.common.bytes.BytesReference; -import org.elasticsearch.common.io.stream.BytesStreamOutput; -import org.elasticsearch.common.io.stream.NamedWriteableAwareStreamInput; -import org.elasticsearch.common.io.stream.StreamInput; -import org.elasticsearch.common.io.stream.Writeable; -import org.elasticsearch.common.lucene.uid.Versions; -import org.elasticsearch.common.settings.SecureString; -import org.elasticsearch.common.util.Maps; -import org.elasticsearch.core.TimeValue; -import org.elasticsearch.index.reindex.AbstractBulkByScrollRequest; -import org.elasticsearch.index.reindex.AbstractBulkIndexByScrollRequest; -import org.elasticsearch.index.reindex.DeleteByQueryRequest; -import org.elasticsearch.index.reindex.ReindexAction; -import org.elasticsearch.index.reindex.ReindexRequest; -import org.elasticsearch.index.reindex.RemoteInfo; -import org.elasticsearch.index.reindex.UpdateByQueryAction; -import org.elasticsearch.index.reindex.UpdateByQueryRequest; -import org.elasticsearch.script.Script; -import org.elasticsearch.script.ScriptType; -import org.elasticsearch.tasks.TaskId; -import org.elasticsearch.test.ESTestCase; -import org.elasticsearch.test.TransportVersionUtils; - -import java.io.IOException; -import java.util.Collections; -import java.util.Map; - -import static org.apache.lucene.tests.util.TestUtil.randomSimpleString; -import static org.elasticsearch.common.bytes.BytesReferenceTestUtils.equalBytes; - -/** - * Round trip tests for all {@link Writeable} things declared in this plugin. - */ -public class RoundTripTests extends ESTestCase { - public void testReindexRequest() throws IOException { - ReindexRequest reindex = new ReindexRequest(); - randomRequest(reindex); - reindex.getDestination().version(randomFrom(Versions.MATCH_ANY, Versions.MATCH_DELETED, 12L, 1L, 123124L, 12L)); - reindex.getDestination().index("test"); - if (randomBoolean()) { - int port = between(1, Integer.MAX_VALUE); - BytesReference query = new BytesArray("{\"match_all\":{}}"); - String username = randomBoolean() ? randomAlphaOfLength(5) : null; - SecureString password = username != null && randomBoolean() ? new SecureString(randomAlphaOfLength(5).toCharArray()) : null; - int headersCount = randomBoolean() ? 0 : between(1, 10); - Map headers = Maps.newMapWithExpectedSize(headersCount); - while (headers.size() < headersCount) { - headers.put(randomAlphaOfLength(5), randomAlphaOfLength(5)); - } - TimeValue socketTimeout = randomPositiveTimeValue(); - TimeValue connectTimeout = randomPositiveTimeValue(); - reindex.setRemoteInfo( - new RemoteInfo( - randomAlphaOfLength(5), - randomAlphaOfLength(5), - port, - null, - query, - username, - password, - headers, - socketTimeout, - connectTimeout - ) - ); - } - ReindexRequest tripped = new ReindexRequest(toInputByteStream(reindex)); - assertRequestEquals(reindex, tripped); - - // Try regular slices with a version that doesn't support slices=auto, which should succeed - reindex.setSlices(between(1, Integer.MAX_VALUE)); - tripped = new ReindexRequest(toInputByteStream(reindex)); - assertRequestEquals(reindex, tripped); - } - - public void testUpdateByQueryRequest() throws IOException { - UpdateByQueryRequest update = new UpdateByQueryRequest(); - randomRequest(update); - if (randomBoolean()) { - update.setPipeline(randomAlphaOfLength(5)); - } - UpdateByQueryRequest tripped = new UpdateByQueryRequest(toInputByteStream(update)); - assertRequestEquals(update, tripped); - assertEquals(update.getPipeline(), tripped.getPipeline()); - - // Try regular slices with a version that doesn't support slices=auto, which should succeed - update.setSlices(between(1, Integer.MAX_VALUE)); - tripped = new UpdateByQueryRequest(toInputByteStream(update)); - assertRequestEquals(update, tripped); - assertEquals(update.getPipeline(), tripped.getPipeline()); - } - - public void testDeleteByQueryRequest() throws IOException { - DeleteByQueryRequest delete = new DeleteByQueryRequest(); - randomRequest(delete); - DeleteByQueryRequest tripped = new DeleteByQueryRequest(toInputByteStream(delete)); - assertRequestEquals(delete, tripped); - - // Try regular slices with a version that doesn't support slices=auto, which should succeed - delete.setSlices(between(1, Integer.MAX_VALUE)); - tripped = new DeleteByQueryRequest(toInputByteStream(delete)); - assertRequestEquals(delete, tripped); - } - - private void randomRequest(AbstractBulkByScrollRequest request) { - request.getSearchRequest().indices("test"); - request.getSearchRequest().source().size(between(1, 1000)); - if (randomBoolean()) { - request.setMaxDocs(between(1, Integer.MAX_VALUE)); - } - request.setAbortOnVersionConflict(random().nextBoolean()); - request.setRefresh(rarely()); - request.setTimeout(randomTimeValue()); - request.setWaitForActiveShards(randomIntBetween(0, 10)); - request.setRequestsPerSecond(between(0, Integer.MAX_VALUE)); - - int slices = ReindexTestCase.randomSlices(1, Integer.MAX_VALUE); - request.setSlices(slices); - - if (randomBoolean()) { - request.setSourceIndicesForDescription(new String[] { "idx1", "idx2" }); - request.getSearchRequest().indices(Strings.EMPTY_ARRAY); - } - } - - private void randomRequest(AbstractBulkIndexByScrollRequest request) { - randomRequest((AbstractBulkByScrollRequest) request); - request.setScript(random().nextBoolean() ? null : randomScript()); - } - - private void assertRequestEquals(ReindexRequest request, ReindexRequest tripped) { - assertRequestEquals((AbstractBulkIndexByScrollRequest) request, (AbstractBulkIndexByScrollRequest) tripped); - assertEquals(request.getDestination().version(), tripped.getDestination().version()); - assertEquals(request.getDestination().index(), tripped.getDestination().index()); - if (request.getRemoteInfo() == null) { - assertNull(tripped.getRemoteInfo()); - } else { - assertNotNull(tripped.getRemoteInfo()); - assertEquals(request.getRemoteInfo().getScheme(), tripped.getRemoteInfo().getScheme()); - assertEquals(request.getRemoteInfo().getHost(), tripped.getRemoteInfo().getHost()); - assertThat(tripped.getRemoteInfo().getQuery(), equalBytes(request.getRemoteInfo().getQuery())); - assertEquals(request.getRemoteInfo().getUsername(), tripped.getRemoteInfo().getUsername()); - assertEquals(request.getRemoteInfo().getPassword(), tripped.getRemoteInfo().getPassword()); - assertEquals(request.getRemoteInfo().getHeaders(), tripped.getRemoteInfo().getHeaders()); - assertEquals(request.getRemoteInfo().getSocketTimeout(), tripped.getRemoteInfo().getSocketTimeout()); - assertEquals(request.getRemoteInfo().getConnectTimeout(), tripped.getRemoteInfo().getConnectTimeout()); - } - } - - private void assertRequestEquals(AbstractBulkIndexByScrollRequest request, AbstractBulkIndexByScrollRequest tripped) { - assertRequestEquals((AbstractBulkByScrollRequest) request, (AbstractBulkByScrollRequest) tripped); - assertEquals(request.getScript(), tripped.getScript()); - } - - private void assertRequestEquals(AbstractBulkByScrollRequest request, AbstractBulkByScrollRequest tripped) { - assertArrayEquals(request.getSearchRequest().indices(), tripped.getSearchRequest().indices()); - assertEquals(request.getSearchRequest().source().size(), tripped.getSearchRequest().source().size()); - assertEquals(request.isAbortOnVersionConflict(), tripped.isAbortOnVersionConflict()); - assertEquals(request.isRefresh(), tripped.isRefresh()); - assertEquals(request.getTimeout(), tripped.getTimeout()); - assertEquals(request.getWaitForActiveShards(), tripped.getWaitForActiveShards()); - assertEquals(request.getRetryBackoffInitialTime(), tripped.getRetryBackoffInitialTime()); - assertEquals(request.getMaxRetries(), tripped.getMaxRetries()); - assertEquals(request.getRequestsPerSecond(), tripped.getRequestsPerSecond(), 0d); - assertArrayEquals( - "sourceIndicesForDescription should round-trip", - request.getSourceIndicesForDescription(), - tripped.getSourceIndicesForDescription() - ); - } - - /** - * Verifies backward compatibility: when reading a ReindexRequest from a node version that predates - * sourceIndicesForDescription, the field is null. - */ - public void testReindexRequestSourceIndicesForDescriptionBwc() throws IOException { - ReindexRequest reindex = new ReindexRequest(); - reindex.getSearchRequest().indices(Strings.EMPTY_ARRAY); - reindex.setSourceIndicesForDescription(new String[] { "source_idx" }); - reindex.setDestIndex("dest"); - reindex.getSearchRequest().source().size(100); - - TransportVersion sourceIndicesForDescriptionVersion = TransportVersion.fromName("bulk_by_scroll_source_indices_for_description"); - TransportVersion versionBeforeFeature = TransportVersionUtils.getPreviousVersion(sourceIndicesForDescriptionVersion); - assumeTrue( - "minimumCompatible must not support sourceIndicesForDescription for this BWC test", - versionBeforeFeature.supports(sourceIndicesForDescriptionVersion) == false - ); - - ReindexRequest tripped = new ReindexRequest(toInputByteStream(versionBeforeFeature, reindex)); - assertNull( - "sourceIndicesForDescription should be null when reading from version before the feature", - tripped.getSourceIndicesForDescription() - ); - } - - public void testRethrottleRequest() throws IOException { - RethrottleRequest request = new RethrottleRequest(); - request.setRequestsPerSecond((float) randomDoubleBetween(0, Float.POSITIVE_INFINITY, false)); - if (randomBoolean()) { - request.setActions(randomFrom(UpdateByQueryAction.NAME, ReindexAction.NAME)); - } else { - request.setTargetTaskId(new TaskId(randomAlphaOfLength(5), randomLong())); - } - RethrottleRequest tripped = new RethrottleRequest(toInputByteStream(request)); - assertEquals(request.getRequestsPerSecond(), tripped.getRequestsPerSecond(), 0.00001); - assertArrayEquals(request.getActions(), tripped.getActions()); - assertEquals(request.getTargetTaskId(), tripped.getTargetTaskId()); - } - - private StreamInput toInputByteStream(Writeable example) throws IOException { - return toInputByteStream(TransportVersion.current(), example); - } - - private StreamInput toInputByteStream(TransportVersion version, Writeable example) throws IOException { - BytesStreamOutput out = new BytesStreamOutput(); - out.setTransportVersion(version); - example.writeTo(out); - StreamInput in = out.bytes().streamInput(); - in.setTransportVersion(version); - return new NamedWriteableAwareStreamInput(in, writableRegistry()); - } - - private Script randomScript() { - ScriptType type = randomFrom(ScriptType.values()); - String lang = random().nextBoolean() ? Script.DEFAULT_SCRIPT_LANG : randomSimpleString(random()); - String idOrCode = randomSimpleString(random()); - Map params = Collections.emptyMap(); - - type = ScriptType.STORED; - - return new Script(type, type == ScriptType.STORED ? null : lang, idOrCode, params); - } -} diff --git a/server/src/main/java/org/elasticsearch/index/reindex/RemoteInfo.java b/server/src/main/java/org/elasticsearch/index/reindex/RemoteInfo.java index 352bbc996fa7e..05f19d812159f 100644 --- a/server/src/main/java/org/elasticsearch/index/reindex/RemoteInfo.java +++ b/server/src/main/java/org/elasticsearch/index/reindex/RemoteInfo.java @@ -133,7 +133,9 @@ public void writeTo(StreamOutput out) throws IOException { @Override public void close() throws IOException { - this.password.close(); + if (password != null) { + password.close(); + } } public String getScheme() { diff --git a/server/src/test/java/org/elasticsearch/index/reindex/BulkByScrollWireSerializingTestUtils.java b/server/src/test/java/org/elasticsearch/index/reindex/BulkByScrollWireSerializingTestUtils.java new file mode 100644 index 0000000000000..531bc0947d3f9 --- /dev/null +++ b/server/src/test/java/org/elasticsearch/index/reindex/BulkByScrollWireSerializingTestUtils.java @@ -0,0 +1,382 @@ +/* + * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one + * or more contributor license agreements. Licensed under the "Elastic License + * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side + * Public License v 1"; you may not use this file except in compliance with, at + * your election, the "Elastic License 2.0", the "GNU Affero General Public + * License v3.0 only", or the "Server Side Public License, v 1". + */ + +package org.elasticsearch.index.reindex; + +import org.elasticsearch.action.support.ActiveShardCount; +import org.elasticsearch.cluster.ClusterModule; +import org.elasticsearch.common.bytes.BytesReference; +import org.elasticsearch.common.io.stream.NamedWriteableRegistry; +import org.elasticsearch.common.settings.SecureString; +import org.elasticsearch.common.settings.Settings; +import org.elasticsearch.search.SearchModule; +import org.elasticsearch.tasks.Task; +import org.elasticsearch.tasks.TaskId; +import org.elasticsearch.test.ESTestCase; +import org.elasticsearch.xcontent.ToXContent; +import org.elasticsearch.xcontent.XContentBuilder; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; + +import static org.elasticsearch.index.query.QueryBuilders.matchAllQuery; +import static org.elasticsearch.index.reindex.resumeinfo.PitWorkerResumeInfoWireSerializingTests.randomPitWorkerResumeInfo; +import static org.elasticsearch.index.reindex.resumeinfo.ScrollWorkerResumeInfoWireSerializingTests.randomScrollWorkerResumeInfo; +import static org.elasticsearch.index.reindex.resumeinfo.SliceStatusWireSerializingTests.sliceMapContentHashCode; +import static org.elasticsearch.index.reindex.resumeinfo.SliceStatusWireSerializingTests.sliceMapsContentEqual; +import static org.elasticsearch.index.reindex.resumeinfo.SliceStatusWireSerializingTests.workerResumeInfoContentEquals; +import static org.elasticsearch.index.reindex.resumeinfo.SliceStatusWireSerializingTests.workerResumeInfoContentHashCode; +import static org.elasticsearch.xcontent.json.JsonXContent.jsonXContent; + +/** + * Shared helpers for bulk-by-scroll wire serialization tests. + */ +public final class BulkByScrollWireSerializingTestUtils { + + private BulkByScrollWireSerializingTestUtils() {} + + /** + * Registry sufficient for {@link ReindexRequest}, {@link UpdateByQueryRequest}, {@link DeleteByQueryRequest}, + * and {@link ResumeBulkByScrollRequest} when the delegate carries {@link ResumeInfo}. + */ + public static NamedWriteableRegistry bulkScrollRequestNamedWriteableRegistry() { + List entries = new ArrayList<>(); + SearchModule searchModule = new SearchModule(Settings.EMPTY, Collections.emptyList()); + entries.addAll(searchModule.getNamedWriteables()); + entries.addAll(ClusterModule.getNamedWriteables()); + entries.add(new NamedWriteableRegistry.Entry(Task.Status.class, BulkByScrollTask.Status.NAME, BulkByScrollTask.Status::new)); + entries.add( + new NamedWriteableRegistry.Entry( + ResumeInfo.WorkerResumeInfo.class, + ResumeInfo.ScrollWorkerResumeInfo.NAME, + ResumeInfo.ScrollWorkerResumeInfo::new + ) + ); + entries.add( + new NamedWriteableRegistry.Entry( + ResumeInfo.WorkerResumeInfo.class, + ResumeInfo.PitWorkerResumeInfo.NAME, + ResumeInfo.PitWorkerResumeInfo::new + ) + ); + return new NamedWriteableRegistry(entries); + } + + public static boolean abstractBulkByScrollRequestsEqual( + AbstractBulkByScrollRequest firstRequest, + AbstractBulkByScrollRequest secondRequest + ) { + if (Objects.equals(firstRequest.getSearchRequest(), secondRequest.getSearchRequest()) == false) { + return false; + } + if (firstRequest.getMaxDocs() != secondRequest.getMaxDocs()) { + return false; + } + if (firstRequest.isAbortOnVersionConflict() != secondRequest.isAbortOnVersionConflict()) { + return false; + } + if (firstRequest.isRefresh() != secondRequest.isRefresh()) { + return false; + } + if (Objects.equals(firstRequest.getTimeout(), secondRequest.getTimeout()) == false) { + return false; + } + if (Objects.equals(firstRequest.getWaitForActiveShards(), secondRequest.getWaitForActiveShards()) == false) { + return false; + } + if (Objects.equals(firstRequest.getRetryBackoffInitialTime(), secondRequest.getRetryBackoffInitialTime()) == false) { + return false; + } + if (firstRequest.getMaxRetries() != secondRequest.getMaxRetries()) { + return false; + } + if (Float.compare(firstRequest.getRequestsPerSecond(), secondRequest.getRequestsPerSecond()) != 0) { + return false; + } + if (firstRequest.getSlices() != secondRequest.getSlices()) { + return false; + } + if (firstRequest.getShouldStoreResult() != secondRequest.getShouldStoreResult()) { + return false; + } + if (firstRequest.isEligibleForRelocationOnShutdown() != secondRequest.isEligibleForRelocationOnShutdown()) { + return false; + } + if (resumeInfosEqual(firstRequest.getResumeInfo(), secondRequest.getResumeInfo()) == false) { + return false; + } + return arraysEqual(firstRequest.getSourceIndicesForDescription(), secondRequest.getSourceIndicesForDescription()); + } + + /** + * Equality for {@link AbstractBulkByScrollRequest#getResumeInfo()} suitable for wire tests: {@link ResumeInfo.PitWorkerResumeInfo} + * uses {@code searchAfterValues} that may change numeric types across {@link org.elasticsearch.common.io.stream.StreamOutput} / + * {@link org.elasticsearch.common.io.stream.StreamInput} round-trips. + */ + public static boolean resumeInfosEqual(Optional first, Optional second) { + if (first.isPresent() != second.isPresent()) { + return false; + } + return first.isEmpty() || resumeInfoContentEquals(first.get(), second.get()); + } + + /** + * Hash code consistent with {@link #resumeInfosEqual(Optional, Optional)} for wire-test wrappers. + */ + public static int resumeInfoOptionalContentHashCode(Optional resumeInfo) { + return resumeInfo.isEmpty() ? 0 : resumeInfoContentHashCode(resumeInfo.get()); + } + + private static boolean resumeInfoContentEquals(ResumeInfo first, ResumeInfo second) { + if (Objects.equals(first.relocationOrigin(), second.relocationOrigin()) == false) { + return false; + } + if (Objects.equals(first.sourceTaskResult(), second.sourceTaskResult()) == false) { + return false; + } + if (first.worker() != null) { + return second.worker() != null + && first.slices() == null + && second.slices() == null + && workerResumeInfoContentEquals(first.worker(), second.worker()); + } + if (first.slices() != null) { + return second.slices() != null && sliceMapsContentEqual(first.slices(), second.slices()); + } + return second.worker() == null && second.slices() == null; + } + + private static int resumeInfoContentHashCode(ResumeInfo resumeInfo) { + int result = Objects.hashCode(resumeInfo.relocationOrigin()); + result = 31 * result + Objects.hashCode(resumeInfo.sourceTaskResult()); + if (resumeInfo.worker() != null) { + result = 31 * result + workerResumeInfoContentHashCode(resumeInfo.worker()); + } else if (resumeInfo.slices() != null) { + result = 31 * result + sliceMapContentHashCode(resumeInfo.slices()); + } + return result; + } + + private static boolean arraysEqual(String[] firstIndices, String[] secondIndices) { + if (firstIndices == null && secondIndices == null) { + return true; + } + if (firstIndices == null || secondIndices == null) { + return false; + } + if (firstIndices.length != secondIndices.length) { + return false; + } + for (int index = 0; index < firstIndices.length; index++) { + if (Objects.equals(firstIndices[index], secondIndices[index]) == false) { + return false; + } + } + return true; + } + + public static boolean abstractBulkIndexByScrollRequestsEqual( + AbstractBulkIndexByScrollRequest firstRequest, + AbstractBulkIndexByScrollRequest secondRequest + ) { + if (abstractBulkByScrollRequestsEqual(firstRequest, secondRequest) == false) { + return false; + } + return Objects.equals(firstRequest.getScript(), secondRequest.getScript()); + } + + public static boolean reindexRequestsEqual(ReindexRequest firstRequest, ReindexRequest secondRequest) { + if (abstractBulkIndexByScrollRequestsEqual(firstRequest, secondRequest) == false) { + return false; + } + if (Objects.equals(firstRequest.getDestination().index(), secondRequest.getDestination().index()) == false) { + return false; + } + if (firstRequest.getDestination().version() != secondRequest.getDestination().version()) { + return false; + } + if (Objects.equals(firstRequest.getRemoteInfo(), secondRequest.getRemoteInfo()) == false) { + return false; + } + return true; + } + + public static boolean updateByQueryRequestsEqual(UpdateByQueryRequest firstRequest, UpdateByQueryRequest secondRequest) { + if (abstractBulkIndexByScrollRequestsEqual(firstRequest, secondRequest) == false) { + return false; + } + return Objects.equals(firstRequest.getPipeline(), secondRequest.getPipeline()); + } + + public static boolean deleteByQueryRequestsEqual(DeleteByQueryRequest firstRequest, DeleteByQueryRequest secondRequest) { + return abstractBulkByScrollRequestsEqual(firstRequest, secondRequest); + } + + public static BytesReference matchAllQueryBytes() { + try { + return BytesReference.bytes(matchAllQuery().toXContent(XContentBuilder.builder(jsonXContent), ToXContent.EMPTY_PARAMS)); + } catch (Exception failure) { + throw new AssertionError(failure); + } + } + + /** + * Random {@link RemoteInfo} with valid JSON query bytes. + */ + public static RemoteInfo randomRemoteInfo() { + BytesReference query = matchAllQueryBytes(); + return new RemoteInfo( + ESTestCase.randomAlphaOfLength(5), + ESTestCase.randomAlphaOfLength(5), + ESTestCase.between(1, 65535), + ESTestCase.randomBoolean() ? ESTestCase.randomAlphaOfLength(4) : null, + query, + ESTestCase.randomBoolean() ? ESTestCase.randomAlphaOfLength(5) : null, + ESTestCase.randomBoolean() ? new SecureString(ESTestCase.randomAlphaOfLength(5).toCharArray()) : null, + ESTestCase.randomBoolean() + ? Map.of(ESTestCase.randomAlphaOfLength(3), ESTestCase.randomAlphaOfLength(3)) + : Collections.emptyMap(), + ESTestCase.randomTimeValue(), + ESTestCase.randomTimeValue() + ); + } + + /** + * Minimal random resume info for embedding in bulk-by-scroll requests (worker or multi-slice). + */ + public static void fillRandomBulkFields(AbstractBulkByScrollRequest request) { + if (ESTestCase.randomBoolean()) { + request.setMaxDocs(ESTestCase.between(100, 10000)); + } + // Else leave maxDocs at MAX_DOCS_ALL_MATCHES (-1): setMaxDocs rejects negatives, so "unlimited" is only the default field value. + request.setAbortOnVersionConflict(ESTestCase.randomBoolean()); + request.setRefresh(ESTestCase.rarely()); + request.setTimeout(ESTestCase.randomTimeValue()); + request.setWaitForActiveShards( + ESTestCase.randomFrom(ActiveShardCount.ALL, ActiveShardCount.NONE, ActiveShardCount.ONE, ActiveShardCount.DEFAULT) + ); + request.setRetryBackoffInitialTime(ESTestCase.randomPositiveTimeValue()); + request.setMaxRetries(ESTestCase.between(0, 20)); + request.setRequestsPerSecond( + ESTestCase.randomBoolean() ? Float.POSITIVE_INFINITY : ESTestCase.randomFloatBetween(0.001f, 1000f, true) + ); + request.setSlices(ESTestCase.between(1, 50)); + request.setShouldStoreResult(ESTestCase.randomBoolean()); + request.setEligibleForRelocationOnShutdown(ESTestCase.randomBoolean()); + if (ESTestCase.randomBoolean()) { + request.setSourceIndicesForDescription(new String[] { "idx1", "idx2" }); + } + if (request.getMaxDocs() != AbstractBulkByScrollRequest.MAX_DOCS_ALL_MATCHES && request.getMaxDocs() < request.getSlices()) { + request.setMaxDocs(request.getSlices()); + } + } + + /** + * Mutates {@code mutatedRequest} (a copy of {@code originalRequest}) by changing exactly one serialized field of + * {@link AbstractBulkByScrollRequest}. + */ + public static void mutateAbstractBulkByScrollRequest( + AbstractBulkByScrollRequest originalRequest, + AbstractBulkByScrollRequest mutatedRequest + ) { + switch (ESTestCase.between(0, 14)) { + case 0 -> { + do { + mutatedRequest.getSearchRequest().indices(ESTestCase.randomAlphaOfLength(12)); + } while (Arrays.equals(originalRequest.getSearchRequest().indices(), mutatedRequest.getSearchRequest().indices())); + } + case 1 -> mutatedRequest.getSearchRequest() + .source() + .size( + ESTestCase.randomValueOtherThan(originalRequest.getSearchRequest().source().size(), () -> ESTestCase.between(1, 2000)) + ); + case 2 -> mutatedRequest.setAbortOnVersionConflict(originalRequest.isAbortOnVersionConflict() == false); + case 3 -> mutatedRequest.setMaxDocs( + ESTestCase.randomValueOtherThan(originalRequest.getMaxDocs(), () -> ESTestCase.between(10, 50000)) + ); + case 4 -> mutatedRequest.setRefresh(originalRequest.isRefresh() == false); + case 5 -> mutatedRequest.setTimeout(ESTestCase.randomValueOtherThan(originalRequest.getTimeout(), ESTestCase::randomTimeValue)); + case 6 -> mutatedRequest.setWaitForActiveShards( + ESTestCase.randomValueOtherThan( + originalRequest.getWaitForActiveShards(), + () -> ESTestCase.randomFrom(ActiveShardCount.ALL, ActiveShardCount.NONE, ActiveShardCount.ONE, ActiveShardCount.DEFAULT) + ) + ); + case 7 -> mutatedRequest.setRetryBackoffInitialTime( + ESTestCase.randomValueOtherThan(originalRequest.getRetryBackoffInitialTime(), ESTestCase::randomPositiveTimeValue) + ); + case 8 -> mutatedRequest.setMaxRetries( + ESTestCase.randomValueOtherThan(originalRequest.getMaxRetries(), () -> ESTestCase.between(0, 50)) + ); + case 9 -> mutatedRequest.setRequestsPerSecond( + ESTestCase.randomValueOtherThan( + originalRequest.getRequestsPerSecond(), + () -> ESTestCase.randomFloatBetween(0.001f, 2000f, true) + ) + ); + case 10 -> mutatedRequest.setSlices( + ESTestCase.randomValueOtherThan(originalRequest.getSlices(), () -> ESTestCase.between(1, 100)) + ); + case 11 -> mutatedRequest.setShouldStoreResult(originalRequest.getShouldStoreResult() == false); + case 12 -> mutatedRequest.setEligibleForRelocationOnShutdown(originalRequest.isEligibleForRelocationOnShutdown() == false); + case 13 -> mutatedRequest.setResumeInfo( + ESTestCase.randomValueOtherThan( + originalRequest.getResumeInfo().orElse(null), + BulkByScrollWireSerializingTestUtils::randomResumeInfo + ) + ); + case 14 -> { + String[] originalIndices = originalRequest.getSourceIndicesForDescription(); + if (originalIndices == null) { + mutatedRequest.setSourceIndicesForDescription(new String[] { ESTestCase.randomAlphaOfLength(6) }); + } else { + do { + mutatedRequest.setSourceIndicesForDescription( + new String[] { ESTestCase.randomAlphaOfLength(9), ESTestCase.randomAlphaOfLength(9) } + ); + } while (Arrays.equals(originalIndices, mutatedRequest.getSourceIndicesForDescription())); + } + } + default -> throw new AssertionError(); + } + if (mutatedRequest.getMaxDocs() != AbstractBulkByScrollRequest.MAX_DOCS_ALL_MATCHES + && mutatedRequest.getMaxDocs() < mutatedRequest.getSlices()) { + mutatedRequest.setMaxDocs(mutatedRequest.getSlices()); + } + } + + /** + * Random {@link ResumeInfo}, either a single worker or multiple slices. All workers in the instance are either + * scroll-based or PIT-based, not a mix. + */ + public static ResumeInfo randomResumeInfo() { + ResumeInfo.RelocationOrigin origin = new ResumeInfo.RelocationOrigin( + new TaskId(ESTestCase.randomAlphaOfLength(8), ESTestCase.randomNonNegativeLong()), + ESTestCase.randomNonNegativeLong() + ); + boolean pitWorkers = ESTestCase.randomBoolean(); + if (ESTestCase.randomBoolean()) { + ResumeInfo.WorkerResumeInfo worker = pitWorkers ? randomPitWorkerResumeInfo() : randomScrollWorkerResumeInfo(); + return new ResumeInfo(origin, worker, null); + } + int sliceCount = ESTestCase.randomIntBetween(2, 4); + HashMap slices = new HashMap<>(); + for (int sliceIndex = 0; sliceIndex < sliceCount; sliceIndex++) { + ResumeInfo.WorkerResumeInfo worker = pitWorkers ? randomPitWorkerResumeInfo() : randomScrollWorkerResumeInfo(); + slices.put(sliceIndex, new ResumeInfo.SliceStatus(sliceIndex, worker, null)); + } + return new ResumeInfo(origin, null, slices); + } +} diff --git a/server/src/test/java/org/elasticsearch/index/reindex/DeleteByQueryRequestWireSerializingTests.java b/server/src/test/java/org/elasticsearch/index/reindex/DeleteByQueryRequestWireSerializingTests.java new file mode 100644 index 0000000000000..6fa7b24283b24 --- /dev/null +++ b/server/src/test/java/org/elasticsearch/index/reindex/DeleteByQueryRequestWireSerializingTests.java @@ -0,0 +1,110 @@ +/* + * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one + * or more contributor license agreements. Licensed under the "Elastic License + * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side + * Public License v 1"; you may not use this file except in compliance with, at + * your election, the "Elastic License 2.0", the "GNU Affero General Public + * License v3.0 only", or the "Server Side Public License, v 1". + */ + +package org.elasticsearch.index.reindex; + +import org.elasticsearch.common.io.stream.BytesStreamOutput; +import org.elasticsearch.common.io.stream.NamedWriteableRegistry; +import org.elasticsearch.common.io.stream.StreamInput; +import org.elasticsearch.common.io.stream.StreamOutput; +import org.elasticsearch.common.io.stream.Writeable; +import org.elasticsearch.test.AbstractWireSerializingTestCase; + +import java.io.IOException; +import java.io.UncheckedIOException; + +import static org.elasticsearch.index.reindex.BulkByScrollWireSerializingTestUtils.fillRandomBulkFields; +import static org.elasticsearch.index.reindex.BulkByScrollWireSerializingTestUtils.mutateAbstractBulkByScrollRequest; +import static org.elasticsearch.index.reindex.BulkByScrollWireSerializingTestUtils.randomResumeInfo; + +public class DeleteByQueryRequestWireSerializingTests extends AbstractWireSerializingTestCase< + DeleteByQueryRequestWireSerializingTests.Wrapper> { + + @Override + protected NamedWriteableRegistry getNamedWriteableRegistry() { + return BulkByScrollWireSerializingTestUtils.bulkScrollRequestNamedWriteableRegistry(); + } + + @Override + protected Writeable.Reader instanceReader() { + return Wrapper::new; + } + + @Override + protected Wrapper createTestInstance() { + DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest(); + deleteByQueryRequest.indices("test-index"); + deleteByQueryRequest.getSearchRequest().source().size(between(1, 1000)); + fillRandomBulkFields(deleteByQueryRequest); + if (randomBoolean()) { + deleteByQueryRequest.setResumeInfo(randomResumeInfo()); + } + return new Wrapper(deleteByQueryRequest); + } + + @Override + protected Wrapper mutateInstance(Wrapper instance) throws IOException { + DeleteByQueryRequest originalRequest = instance.request; + DeleteByQueryRequest mutatedRequest = copyInstance(instance).request; + mutateAbstractBulkByScrollRequest(originalRequest, mutatedRequest); + return new Wrapper(mutatedRequest); + } + + static final class Wrapper implements Writeable { + private final DeleteByQueryRequest request; + + Wrapper(DeleteByQueryRequest request) { + this.request = request; + } + + Wrapper(StreamInput streamInput) throws IOException { + this.request = new DeleteByQueryRequest(streamInput); + } + + @Override + public void writeTo(StreamOutput output) throws IOException { + request.writeTo(output); + } + + @Override + public boolean equals(Object other) { + if (this == other) return true; + if (other == null || getClass() != other.getClass()) return false; + Wrapper wrapper = (Wrapper) other; + return wireBytesEqual(request, wrapper.request); + } + + @Override + public int hashCode() { + try { + BytesStreamOutput output = new BytesStreamOutput(); + request.writeTo(output); + return output.bytes().hashCode(); + } catch (IOException ioException) { + throw new UncheckedIOException(ioException); + } + } + + /** + * Equality by serialized wire form. Deep {@link DeleteByQueryRequest} equality can disagree with a transport round-trip + * (e.g. nested resume state) even when the encoded bytes match; wire tests should assert encoding stability. + */ + private static boolean wireBytesEqual(DeleteByQueryRequest firstRequest, DeleteByQueryRequest secondRequest) { + try { + BytesStreamOutput firstOutput = new BytesStreamOutput(); + firstRequest.writeTo(firstOutput); + BytesStreamOutput secondOutput = new BytesStreamOutput(); + secondRequest.writeTo(secondOutput); + return firstOutput.bytes().equals(secondOutput.bytes()); + } catch (IOException ioException) { + throw new UncheckedIOException(ioException); + } + } + } +} diff --git a/server/src/test/java/org/elasticsearch/index/reindex/ReindexRequestWireSerializingTests.java b/server/src/test/java/org/elasticsearch/index/reindex/ReindexRequestWireSerializingTests.java new file mode 100644 index 0000000000000..bca093f7aa879 --- /dev/null +++ b/server/src/test/java/org/elasticsearch/index/reindex/ReindexRequestWireSerializingTests.java @@ -0,0 +1,193 @@ +/* + * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one + * or more contributor license agreements. Licensed under the "Elastic License + * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side + * Public License v 1"; you may not use this file except in compliance with, at + * your election, the "Elastic License 2.0", the "GNU Affero General Public + * License v3.0 only", or the "Server Side Public License, v 1". + */ + +package org.elasticsearch.index.reindex; + +import org.elasticsearch.TransportVersion; +import org.elasticsearch.common.Strings; +import org.elasticsearch.common.io.stream.NamedWriteableRegistry; +import org.elasticsearch.common.io.stream.StreamInput; +import org.elasticsearch.common.io.stream.StreamOutput; +import org.elasticsearch.common.io.stream.Writeable; +import org.elasticsearch.common.lucene.uid.Versions; +import org.elasticsearch.script.Script; +import org.elasticsearch.script.ScriptType; +import org.elasticsearch.test.AbstractWireSerializingTestCase; +import org.elasticsearch.test.ESTestCase; +import org.elasticsearch.test.TransportVersionUtils; + +import java.io.IOException; +import java.util.Arrays; +import java.util.Collections; +import java.util.Objects; + +import static org.elasticsearch.index.reindex.BulkByScrollWireSerializingTestUtils.fillRandomBulkFields; +import static org.elasticsearch.index.reindex.BulkByScrollWireSerializingTestUtils.mutateAbstractBulkByScrollRequest; +import static org.elasticsearch.index.reindex.BulkByScrollWireSerializingTestUtils.randomRemoteInfo; +import static org.elasticsearch.index.reindex.BulkByScrollWireSerializingTestUtils.randomResumeInfo; +import static org.elasticsearch.index.reindex.BulkByScrollWireSerializingTestUtils.reindexRequestsEqual; +import static org.elasticsearch.index.reindex.BulkByScrollWireSerializingTestUtils.resumeInfoOptionalContentHashCode; + +public class ReindexRequestWireSerializingTests extends AbstractWireSerializingTestCase { + + @Override + protected NamedWriteableRegistry getNamedWriteableRegistry() { + return BulkByScrollWireSerializingTestUtils.bulkScrollRequestNamedWriteableRegistry(); + } + + @Override + protected Writeable.Reader instanceReader() { + return Wrapper::new; + } + + @Override + protected Wrapper createTestInstance() { + return new Wrapper(newRandomReindexWireInstance()); + } + + /** + * Random {@link ReindexRequest}. + */ + public static ReindexRequest newRandomReindexWireInstance() { + ReindexRequest reindexRequest = new ReindexRequest(); + reindexRequest.setSourceIndices("source"); + reindexRequest.setDestIndex("destination"); + reindexRequest.getDestination().version(ESTestCase.randomFrom(Versions.MATCH_ANY, Versions.MATCH_DELETED, 12L, 1L, 123124L)); + reindexRequest.getSearchRequest().source().size(ESTestCase.between(1, 1000)); + fillRandomBulkFields(reindexRequest); + if (ESTestCase.randomBoolean()) { + reindexRequest.setScript(new Script(ScriptType.STORED, null, ESTestCase.randomAlphaOfLength(6), Collections.emptyMap())); + } + if (ESTestCase.randomBoolean()) { + reindexRequest.setRemoteInfo(randomRemoteInfo()); + } + if (ESTestCase.randomBoolean()) { + reindexRequest.setResumeInfo(randomResumeInfo()); + } + return reindexRequest; + } + + /** + * When reading a {@link ReindexRequest} from a node version that predates {@code sourceIndicesForDescription}, + * the field is absent on the wire and must deserialize as {@code null}. + */ + public void testReindexRequestSourceIndicesForDescriptionBwc() throws IOException { + ReindexRequest reindex = new ReindexRequest(); + reindex.getSearchRequest().indices(Strings.EMPTY_ARRAY); + reindex.setSourceIndicesForDescription(new String[] { "source_idx" }); + reindex.setDestIndex("dest"); + reindex.getSearchRequest().source().size(100); + + TransportVersion sourceIndicesForDescriptionVersion = TransportVersion.fromName("bulk_by_scroll_source_indices_for_description"); + TransportVersion versionBeforeFeature = TransportVersionUtils.getPreviousVersion(sourceIndicesForDescriptionVersion); + assumeTrue( + "minimumCompatible must not support sourceIndicesForDescription for this BWC test", + versionBeforeFeature.supports(sourceIndicesForDescriptionVersion) == false + ); + + Wrapper tripped = copyInstance(new Wrapper(reindex), versionBeforeFeature); + assertNull( + "sourceIndicesForDescription should be null when reading from version before the feature", + tripped.request.getSourceIndicesForDescription() + ); + } + + @Override + protected Wrapper mutateInstance(Wrapper instance) throws IOException { + ReindexRequest originalRequest = instance.request; + ReindexRequest mutatedRequest = copyInstance(instance).request; + mutateReindexRequest(originalRequest, mutatedRequest); + return new Wrapper(mutatedRequest); + } + + /** + * Mutates {@code mutatedRequest} (a copy of {@code originalRequest}) along exactly one logical field: bulk-by-scroll fields, + * destination index, destination version, remote info, or script. + */ + public static void mutateReindexRequest(ReindexRequest originalRequest, ReindexRequest mutatedRequest) { + switch (between(0, 4)) { + case 0 -> mutateAbstractBulkByScrollRequest(originalRequest, mutatedRequest); + case 1 -> mutatedRequest.getDestination() + .index(ESTestCase.randomValueOtherThan(originalRequest.getDestination().index(), () -> ESTestCase.randomAlphaOfLength(14))); + case 2 -> mutatedRequest.getDestination() + .version( + ESTestCase.randomValueOtherThan( + originalRequest.getDestination().version(), + () -> ESTestCase.randomFrom(Versions.MATCH_ANY, Versions.MATCH_DELETED, 99L) + ) + ); + case 3 -> mutatedRequest.setRemoteInfo( + ESTestCase.randomValueOtherThan( + originalRequest.getRemoteInfo(), + () -> ESTestCase.randomBoolean() ? randomRemoteInfo() : null + ) + ); + case 4 -> mutatedRequest.setScript( + ESTestCase.randomValueOtherThan( + originalRequest.getScript(), + () -> new Script(ScriptType.STORED, null, ESTestCase.randomAlphaOfLength(11), Collections.emptyMap()) + ) + ); + default -> throw new AssertionError(); + } + if (mutatedRequest.getMaxDocs() != AbstractBulkByScrollRequest.MAX_DOCS_ALL_MATCHES + && mutatedRequest.getMaxDocs() < mutatedRequest.getSlices()) { + mutatedRequest.setMaxDocs(mutatedRequest.getSlices()); + } + } + + static final class Wrapper implements Writeable { + private final ReindexRequest request; + + Wrapper(ReindexRequest request) { + this.request = request; + } + + Wrapper(StreamInput streamInput) throws IOException { + this.request = new ReindexRequest(streamInput); + } + + @Override + public void writeTo(StreamOutput output) throws IOException { + request.writeTo(output); + } + + @Override + public boolean equals(Object other) { + if (this == other) return true; + if (other == null || getClass() != other.getClass()) return false; + Wrapper wrapper = (Wrapper) other; + return reindexRequestsEqual(request, wrapper.request); + } + + @Override + public int hashCode() { + return Objects.hash( + request.getSearchRequest(), + request.getMaxDocs(), + request.isAbortOnVersionConflict(), + request.isRefresh(), + request.getTimeout(), + request.getWaitForActiveShards(), + request.getRetryBackoffInitialTime(), + request.getMaxRetries(), + request.getRequestsPerSecond(), + request.getSlices(), + request.getShouldStoreResult(), + request.isEligibleForRelocationOnShutdown(), + resumeInfoOptionalContentHashCode(request.getResumeInfo()), + Arrays.hashCode(request.getSourceIndicesForDescription()), + request.getDestination().index(), + request.getDestination().version(), + request.getRemoteInfo(), + request.getScript() + ); + } + } +} diff --git a/server/src/test/java/org/elasticsearch/index/reindex/RemoteInfoWireSerializingTests.java b/server/src/test/java/org/elasticsearch/index/reindex/RemoteInfoWireSerializingTests.java new file mode 100644 index 0000000000000..aab64c52a9e63 --- /dev/null +++ b/server/src/test/java/org/elasticsearch/index/reindex/RemoteInfoWireSerializingTests.java @@ -0,0 +1,213 @@ +/* + * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one + * or more contributor license agreements. Licensed under the "Elastic License + * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side + * Public License v 1"; you may not use this file except in compliance with, at + * your election, the "Elastic License 2.0", the "GNU Affero General Public + * License v3.0 only", or the "Server Side Public License, v 1". + */ + +package org.elasticsearch.index.reindex; + +import org.elasticsearch.common.bytes.BytesReference; +import org.elasticsearch.common.io.stream.Writeable; +import org.elasticsearch.common.settings.SecureString; +import org.elasticsearch.core.Nullable; +import org.elasticsearch.index.query.QueryBuilders; +import org.elasticsearch.test.AbstractWireSerializingTestCase; +import org.elasticsearch.test.ESTestCase; +import org.elasticsearch.xcontent.ToXContent; +import org.elasticsearch.xcontent.XContentBuilder; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.Map; + +import static org.elasticsearch.index.reindex.BulkByScrollWireSerializingTestUtils.matchAllQueryBytes; +import static org.elasticsearch.index.reindex.BulkByScrollWireSerializingTestUtils.randomRemoteInfo; +import static org.elasticsearch.xcontent.json.JsonXContent.jsonXContent; + +public class RemoteInfoWireSerializingTests extends AbstractWireSerializingTestCase { + + /** + * {@link RemoteInfo} mutations must not share a {@link SecureString} with the instance being mutated: {@link #dispose} closes the + * mutation and would invalidate the original before wire equality checks run. + */ + private static SecureString copyPassword(@Nullable SecureString password) { + return password == null ? null : password.clone(); + } + + /** Valid non-{@link QueryBuilders#matchAllQuery()} JSON for remote query */ + private static BytesReference termQueryBytes() { + try { + return BytesReference.bytes( + QueryBuilders.termQuery("_id", "mutated").toXContent(XContentBuilder.builder(jsonXContent), ToXContent.EMPTY_PARAMS) + ); + } catch (IOException ioException) { + throw new AssertionError(ioException); + } + } + + /** + * {@link BulkByScrollWireSerializingTestUtils#randomRemoteInfo()} always uses + * {@link BulkByScrollWireSerializingTestUtils#matchAllQueryBytes()}; toggling with a different fixed query avoids + * {@link ESTestCase#randomValueOtherThan} spinning forever when the supplier keeps returning the same bytes as the input. + */ + private static BytesReference queryMutation(BytesReference current) { + BytesReference matchAll = matchAllQueryBytes(); + BytesReference term = termQueryBytes(); + return current.equals(matchAll) ? term : matchAll; + } + + @Override + protected Writeable.Reader instanceReader() { + return RemoteInfo::new; + } + + @Override + protected RemoteInfo createTestInstance() { + return randomRemoteInfo(); + } + + @Override + protected RemoteInfo mutateInstance(RemoteInfo instance) throws IOException { + return switch (between(0, 9)) { + case 0 -> new RemoteInfo( + randomValueOtherThan(instance.getScheme(), () -> randomFrom("http", "https")), + instance.getHost(), + instance.getPort(), + instance.getPathPrefix(), + instance.getQuery(), + instance.getUsername(), + copyPassword(instance.getPassword()), + instance.getHeaders(), + instance.getSocketTimeout(), + instance.getConnectTimeout() + ); + case 1 -> new RemoteInfo( + instance.getScheme(), + randomValueOtherThan(instance.getHost(), () -> randomAlphaOfLength(12)), + instance.getPort(), + instance.getPathPrefix(), + instance.getQuery(), + instance.getUsername(), + copyPassword(instance.getPassword()), + instance.getHeaders(), + instance.getSocketTimeout(), + instance.getConnectTimeout() + ); + case 2 -> new RemoteInfo( + instance.getScheme(), + instance.getHost(), + randomValueOtherThan(instance.getPort(), () -> between(1, 65535)), + instance.getPathPrefix(), + instance.getQuery(), + instance.getUsername(), + copyPassword(instance.getPassword()), + instance.getHeaders(), + instance.getSocketTimeout(), + instance.getConnectTimeout() + ); + case 3 -> new RemoteInfo( + instance.getScheme(), + instance.getHost(), + instance.getPort(), + randomValueOtherThan(instance.getPathPrefix(), () -> randomBoolean() ? null : randomAlphaOfLength(6)), + instance.getQuery(), + instance.getUsername(), + copyPassword(instance.getPassword()), + instance.getHeaders(), + instance.getSocketTimeout(), + instance.getConnectTimeout() + ); + case 4 -> new RemoteInfo( + instance.getScheme(), + instance.getHost(), + instance.getPort(), + instance.getPathPrefix(), + queryMutation(instance.getQuery()), + instance.getUsername(), + copyPassword(instance.getPassword()), + instance.getHeaders(), + instance.getSocketTimeout(), + instance.getConnectTimeout() + ); + case 5 -> new RemoteInfo( + instance.getScheme(), + instance.getHost(), + instance.getPort(), + instance.getPathPrefix(), + instance.getQuery(), + randomValueOtherThan(instance.getUsername(), () -> randomBoolean() ? randomAlphaOfLength(5) : null), + copyPassword(instance.getPassword()), + instance.getHeaders(), + instance.getSocketTimeout(), + instance.getConnectTimeout() + ); + case 6 -> new RemoteInfo( + instance.getScheme(), + instance.getHost(), + instance.getPort(), + instance.getPathPrefix(), + instance.getQuery(), + instance.getUsername(), + // SecureString equality is content-based: a clone with the same chars still equals the original. + instance.getPassword() == null + ? new SecureString(randomAlphaOfLength(8).toCharArray()) + : randomValueOtherThan(instance.getPassword(), () -> new SecureString(randomAlphaOfLength(9).toCharArray())), + instance.getHeaders(), + instance.getSocketTimeout(), + instance.getConnectTimeout() + ); + case 7 -> new RemoteInfo( + instance.getScheme(), + instance.getHost(), + instance.getPort(), + instance.getPathPrefix(), + instance.getQuery(), + instance.getUsername(), + copyPassword(instance.getPassword()), + randomValueOtherThan(instance.getHeaders(), () -> Map.of(randomAlphaOfLength(3), randomAlphaOfLength(4))), + instance.getSocketTimeout(), + instance.getConnectTimeout() + ); + case 8 -> new RemoteInfo( + instance.getScheme(), + instance.getHost(), + instance.getPort(), + instance.getPathPrefix(), + instance.getQuery(), + instance.getUsername(), + copyPassword(instance.getPassword()), + instance.getHeaders(), + randomValueOtherThan(instance.getSocketTimeout(), ESTestCase::randomTimeValue), + instance.getConnectTimeout() + ); + case 9 -> new RemoteInfo( + instance.getScheme(), + instance.getHost(), + instance.getPort(), + instance.getPathPrefix(), + instance.getQuery(), + instance.getUsername(), + copyPassword(instance.getPassword()), + instance.getHeaders(), + instance.getSocketTimeout(), + randomValueOtherThan(instance.getConnectTimeout(), ESTestCase::randomTimeValue) + ); + default -> throw new AssertionError(); + }; + } + + /** + * Releases the {@link RemoteInfo} secure string password rather than relying on GC eventually cleaning it up + */ + @Override + protected void dispose(RemoteInfo remoteInfo) { + try { + remoteInfo.close(); + } catch (IOException ioException) { + throw new UncheckedIOException(ioException); + } + } +} diff --git a/server/src/test/java/org/elasticsearch/index/reindex/ResumeBulkByScrollRequestWireSerializingTests.java b/server/src/test/java/org/elasticsearch/index/reindex/ResumeBulkByScrollRequestWireSerializingTests.java new file mode 100644 index 0000000000000..4f662012e90f9 --- /dev/null +++ b/server/src/test/java/org/elasticsearch/index/reindex/ResumeBulkByScrollRequestWireSerializingTests.java @@ -0,0 +1,104 @@ +/* + * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one + * or more contributor license agreements. Licensed under the "Elastic License + * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side + * Public License v 1"; you may not use this file except in compliance with, at + * your election, the "Elastic License 2.0", the "GNU Affero General Public + * License v3.0 only", or the "Server Side Public License, v 1". + */ + +package org.elasticsearch.index.reindex; + +import org.elasticsearch.common.io.stream.NamedWriteableRegistry; +import org.elasticsearch.common.io.stream.StreamInput; +import org.elasticsearch.common.io.stream.StreamOutput; +import org.elasticsearch.common.io.stream.Writeable; +import org.elasticsearch.test.AbstractWireSerializingTestCase; + +import java.io.IOException; +import java.util.Arrays; +import java.util.Objects; + +import static org.elasticsearch.index.reindex.BulkByScrollWireSerializingTestUtils.reindexRequestsEqual; +import static org.elasticsearch.index.reindex.BulkByScrollWireSerializingTestUtils.resumeInfoOptionalContentHashCode; +import static org.elasticsearch.index.reindex.ReindexRequestWireSerializingTests.mutateReindexRequest; +import static org.elasticsearch.index.reindex.ReindexRequestWireSerializingTests.newRandomReindexWireInstance; + +public class ResumeBulkByScrollRequestWireSerializingTests extends AbstractWireSerializingTestCase< + ResumeBulkByScrollRequestWireSerializingTests.Wrapper> { + + @Override + protected NamedWriteableRegistry getNamedWriteableRegistry() { + return BulkByScrollWireSerializingTestUtils.bulkScrollRequestNamedWriteableRegistry(); + } + + @Override + protected Writeable.Reader instanceReader() { + return Wrapper::new; + } + + @Override + protected Wrapper createTestInstance() { + return new Wrapper(new ResumeBulkByScrollRequest(newRandomReindexWireInstance())); + } + + @Override + protected Wrapper mutateInstance(Wrapper instance) throws IOException { + ResumeBulkByScrollRequest originalRequest = instance.request; + ResumeBulkByScrollRequest mutatedResumeRequest = copyInstance(instance).request; + ReindexRequest originalReindex = (ReindexRequest) originalRequest.getDelegate(); + ReindexRequest mutatedReindex = (ReindexRequest) mutatedResumeRequest.getDelegate(); + mutateReindexRequest(originalReindex, mutatedReindex); + return new Wrapper(new ResumeBulkByScrollRequest(mutatedReindex)); + } + + static final class Wrapper implements Writeable { + private final ResumeBulkByScrollRequest request; + + Wrapper(ResumeBulkByScrollRequest request) { + this.request = request; + } + + Wrapper(StreamInput streamInput) throws IOException { + this.request = new ResumeBulkByScrollRequest(streamInput, ReindexRequest::new); + } + + @Override + public void writeTo(StreamOutput output) throws IOException { + request.writeTo(output); + } + + @Override + public boolean equals(Object other) { + if (this == other) return true; + if (other == null || getClass() != other.getClass()) return false; + Wrapper wrapper = (Wrapper) other; + return reindexRequestsEqual((ReindexRequest) request.getDelegate(), (ReindexRequest) wrapper.request.getDelegate()); + } + + @Override + public int hashCode() { + ReindexRequest reindexDelegate = (ReindexRequest) request.getDelegate(); + return Objects.hash( + reindexDelegate.getSearchRequest(), + reindexDelegate.getMaxDocs(), + reindexDelegate.isAbortOnVersionConflict(), + reindexDelegate.isRefresh(), + reindexDelegate.getTimeout(), + reindexDelegate.getWaitForActiveShards(), + reindexDelegate.getRetryBackoffInitialTime(), + reindexDelegate.getMaxRetries(), + reindexDelegate.getRequestsPerSecond(), + reindexDelegate.getSlices(), + reindexDelegate.getShouldStoreResult(), + reindexDelegate.isEligibleForRelocationOnShutdown(), + resumeInfoOptionalContentHashCode(reindexDelegate.getResumeInfo()), + Arrays.hashCode(reindexDelegate.getSourceIndicesForDescription()), + reindexDelegate.getDestination().index(), + reindexDelegate.getDestination().version(), + reindexDelegate.getRemoteInfo(), + reindexDelegate.getScript() + ); + } + } +} diff --git a/server/src/test/java/org/elasticsearch/index/reindex/ResumeBulkByScrollResponseWireSerializingTests.java b/server/src/test/java/org/elasticsearch/index/reindex/ResumeBulkByScrollResponseWireSerializingTests.java new file mode 100644 index 0000000000000..224ab64be6071 --- /dev/null +++ b/server/src/test/java/org/elasticsearch/index/reindex/ResumeBulkByScrollResponseWireSerializingTests.java @@ -0,0 +1,70 @@ +/* + * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one + * or more contributor license agreements. Licensed under the "Elastic License + * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side + * Public License v 1"; you may not use this file except in compliance with, at + * your election, the "Elastic License 2.0", the "GNU Affero General Public + * License v3.0 only", or the "Server Side Public License, v 1". + */ + +package org.elasticsearch.index.reindex; + +import org.elasticsearch.common.io.stream.StreamInput; +import org.elasticsearch.common.io.stream.StreamOutput; +import org.elasticsearch.common.io.stream.Writeable; +import org.elasticsearch.tasks.TaskId; +import org.elasticsearch.test.AbstractWireSerializingTestCase; + +import java.io.IOException; +import java.util.Objects; + +public class ResumeBulkByScrollResponseWireSerializingTests extends AbstractWireSerializingTestCase< + ResumeBulkByScrollResponseWireSerializingTests.Wrapper> { + + @Override + protected Writeable.Reader instanceReader() { + return Wrapper::new; + } + + @Override + protected Wrapper createTestInstance() { + return new Wrapper(new ResumeBulkByScrollResponse(new TaskId(randomAlphaOfLength(8), randomNonNegativeLong()))); + } + + @Override + protected Wrapper mutateInstance(Wrapper instance) throws IOException { + TaskId origId = instance.response.getTaskId(); + TaskId newId = randomValueOtherThan(origId, () -> new TaskId(randomAlphaOfLength(10), randomNonNegativeLong())); + return new Wrapper(new ResumeBulkByScrollResponse(newId)); + } + + static final class Wrapper implements Writeable { + private final ResumeBulkByScrollResponse response; + + Wrapper(ResumeBulkByScrollResponse response) { + this.response = response; + } + + Wrapper(StreamInput in) throws IOException { + this.response = new ResumeBulkByScrollResponse(in); + } + + @Override + public void writeTo(StreamOutput out) throws IOException { + response.writeTo(out); + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + Wrapper wrapper = (Wrapper) o; + return Objects.equals(response.getTaskId(), wrapper.response.getTaskId()); + } + + @Override + public int hashCode() { + return Objects.hashCode(response.getTaskId()); + } + } +} diff --git a/server/src/test/java/org/elasticsearch/index/reindex/UpdateByQueryRequestWireSerializingTests.java b/server/src/test/java/org/elasticsearch/index/reindex/UpdateByQueryRequestWireSerializingTests.java new file mode 100644 index 0000000000000..144e444ffbdc6 --- /dev/null +++ b/server/src/test/java/org/elasticsearch/index/reindex/UpdateByQueryRequestWireSerializingTests.java @@ -0,0 +1,126 @@ +/* + * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one + * or more contributor license agreements. Licensed under the "Elastic License + * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side + * Public License v 1"; you may not use this file except in compliance with, at + * your election, the "Elastic License 2.0", the "GNU Affero General Public + * License v3.0 only", or the "Server Side Public License, v 1". + */ + +package org.elasticsearch.index.reindex; + +import org.elasticsearch.common.io.stream.NamedWriteableRegistry; +import org.elasticsearch.common.io.stream.StreamInput; +import org.elasticsearch.common.io.stream.StreamOutput; +import org.elasticsearch.common.io.stream.Writeable; +import org.elasticsearch.script.Script; +import org.elasticsearch.script.ScriptType; +import org.elasticsearch.test.AbstractWireSerializingTestCase; + +import java.io.IOException; +import java.util.Arrays; +import java.util.Collections; +import java.util.Objects; + +import static org.elasticsearch.index.reindex.BulkByScrollWireSerializingTestUtils.fillRandomBulkFields; +import static org.elasticsearch.index.reindex.BulkByScrollWireSerializingTestUtils.mutateAbstractBulkByScrollRequest; +import static org.elasticsearch.index.reindex.BulkByScrollWireSerializingTestUtils.randomResumeInfo; +import static org.elasticsearch.index.reindex.BulkByScrollWireSerializingTestUtils.resumeInfoOptionalContentHashCode; +import static org.elasticsearch.index.reindex.BulkByScrollWireSerializingTestUtils.updateByQueryRequestsEqual; + +public class UpdateByQueryRequestWireSerializingTests extends AbstractWireSerializingTestCase< + UpdateByQueryRequestWireSerializingTests.Wrapper> { + + @Override + protected NamedWriteableRegistry getNamedWriteableRegistry() { + return BulkByScrollWireSerializingTestUtils.bulkScrollRequestNamedWriteableRegistry(); + } + + @Override + protected Writeable.Reader instanceReader() { + return Wrapper::new; + } + + @Override + protected Wrapper createTestInstance() { + UpdateByQueryRequest updateByQueryRequest = new UpdateByQueryRequest(); + updateByQueryRequest.indices("test-index"); + updateByQueryRequest.getSearchRequest().source().size(between(1, 1000)); + fillRandomBulkFields(updateByQueryRequest); + if (randomBoolean()) { + updateByQueryRequest.setPipeline(randomAlphaOfLength(5)); + } + if (randomBoolean()) { + updateByQueryRequest.setScript(new Script(ScriptType.STORED, null, randomAlphaOfLength(6), Collections.emptyMap())); + } + if (randomBoolean()) { + updateByQueryRequest.setResumeInfo(randomResumeInfo()); + } + return new Wrapper(updateByQueryRequest); + } + + @Override + protected Wrapper mutateInstance(Wrapper instance) throws IOException { + UpdateByQueryRequest originalRequest = instance.request; + UpdateByQueryRequest mutatedRequest = copyInstance(instance).request; + switch (between(0, 2)) { + case 0 -> mutateAbstractBulkByScrollRequest(originalRequest, mutatedRequest); + case 1 -> mutatedRequest.setScript( + randomValueOtherThan( + originalRequest.getScript(), + () -> new Script(ScriptType.STORED, null, randomAlphaOfLength(11), Collections.emptyMap()) + ) + ); + case 2 -> mutatedRequest.setPipeline(randomValueOtherThan(originalRequest.getPipeline(), () -> randomAlphaOfLength(12))); + default -> throw new AssertionError(); + } + return new Wrapper(mutatedRequest); + } + + static final class Wrapper implements Writeable { + private final UpdateByQueryRequest request; + + Wrapper(UpdateByQueryRequest request) { + this.request = request; + } + + Wrapper(StreamInput streamInput) throws IOException { + this.request = new UpdateByQueryRequest(streamInput); + } + + @Override + public void writeTo(StreamOutput output) throws IOException { + request.writeTo(output); + } + + @Override + public boolean equals(Object other) { + if (this == other) return true; + if (other == null || getClass() != other.getClass()) return false; + Wrapper wrapper = (Wrapper) other; + return updateByQueryRequestsEqual(request, wrapper.request); + } + + @Override + public int hashCode() { + return Objects.hash( + request.getSearchRequest(), + request.getMaxDocs(), + request.isAbortOnVersionConflict(), + request.isRefresh(), + request.getTimeout(), + request.getWaitForActiveShards(), + request.getRetryBackoffInitialTime(), + request.getMaxRetries(), + request.getRequestsPerSecond(), + request.getSlices(), + request.getShouldStoreResult(), + request.isEligibleForRelocationOnShutdown(), + resumeInfoOptionalContentHashCode(request.getResumeInfo()), + Arrays.hashCode(request.getSourceIndicesForDescription()), + request.getScript(), + request.getPipeline() + ); + } + } +} diff --git a/server/src/test/java/org/elasticsearch/index/reindex/resumeinfo/SliceStatusWireSerializingTests.java b/server/src/test/java/org/elasticsearch/index/reindex/resumeinfo/SliceStatusWireSerializingTests.java index d0d2967e2438f..628fe60c0c55b 100644 --- a/server/src/test/java/org/elasticsearch/index/reindex/resumeinfo/SliceStatusWireSerializingTests.java +++ b/server/src/test/java/org/elasticsearch/index/reindex/resumeinfo/SliceStatusWireSerializingTests.java @@ -28,7 +28,9 @@ import java.io.IOException; import java.util.ArrayList; +import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.Objects; import static java.util.Collections.emptyList; @@ -72,7 +74,7 @@ protected Wrapper mutateInstance(Wrapper instance) throws IOException { return null; } - static boolean workerResumeInfoContentEquals(WorkerResumeInfo a, WorkerResumeInfo b) { + public static boolean workerResumeInfoContentEquals(WorkerResumeInfo a, WorkerResumeInfo b) { if (Objects.equals(a, b)) return true; if (a == null || b == null) return false; if (a instanceof PitWorkerResumeInfo pa && b instanceof PitWorkerResumeInfo pb) { @@ -81,7 +83,7 @@ static boolean workerResumeInfoContentEquals(WorkerResumeInfo a, WorkerResumeInf return a.equals(b); } - static int workerResumeInfoContentHashCode(WorkerResumeInfo info) { + public static int workerResumeInfoContentHashCode(WorkerResumeInfo info) { if (info == null) return 0; if (info instanceof PitWorkerResumeInfo pit) { return pitWorkerResumeInfoContentHashCode(pit); @@ -89,7 +91,7 @@ static int workerResumeInfoContentHashCode(WorkerResumeInfo info) { return info.hashCode(); } - static boolean sliceStatusContentEquals(SliceStatus a, SliceStatus b) { + public static boolean sliceStatusContentEquals(SliceStatus a, SliceStatus b) { if (a.sliceId() != b.sliceId()) { return false; } @@ -103,7 +105,7 @@ static boolean sliceStatusContentEquals(SliceStatus a, SliceStatus b) { return workerResultContentEquals(a.result(), b.result()); } - static int sliceStatusContentHashCode(SliceStatus status) { + public static int sliceStatusContentHashCode(SliceStatus status) { int result = Integer.hashCode(status.sliceId()); result = 31 * result + workerResumeInfoContentHashCode(status.resumeInfo()); if (status.result() != null) { @@ -150,6 +152,36 @@ static boolean bulkByScrollResponseContentEquals(BulkByScrollResponse a, BulkByS return true; } + /** + * Compares slice maps (e.g. {@link org.elasticsearch.index.reindex.ResumeInfo#slices()}) for wire-test equality. + */ + public static boolean sliceMapsContentEqual(Map first, Map second) { + if (first.size() != second.size()) { + return false; + } + for (Map.Entry entry : first.entrySet()) { + SliceStatus otherSlice = second.get(entry.getKey()); + if (otherSlice == null || sliceStatusContentEquals(entry.getValue(), otherSlice) == false) { + return false; + } + } + return true; + } + + /** + * Hash code consistent with {@link #sliceMapsContentEqual(Map, Map)} for wire-test wrappers. + */ + public static int sliceMapContentHashCode(Map slices) { + List keys = new ArrayList<>(slices.keySet()); + Collections.sort(keys); + int result = 1; + for (Integer key : keys) { + result = 31 * result + key; + result = 31 * result + sliceStatusContentHashCode(slices.get(key)); + } + return result; + } + /** * Wrapper around {@link SliceStatus} that implements content-based equals/hashCode so that * round-trip serialization tests pass when the slice contains {@link BulkByScrollResponse} or {@link Exception}.