Add BulkByScrollTask Serialization Tests

server/src/test/java/org/elasticsearch/index/reindex/BulkByScrollTaskStatusOrExceptionTests.java

@@ -76,6 +76,52 @@ protected boolean supportsUnknownFields() {
         return true;
     }
 
+    /**
+     * Verifies that a {@link StatusOrException} constructed with a {@link BulkByScrollTask.Status} exposes it via
+     * {@link StatusOrException#getStatus()} and returns null from {@link StatusOrException#getException()}.
+     */
+    public void testStatusOrExceptionWithStatus() {
+        BulkByScrollTask.Status status = BulkByScrollTaskStatusTests.randomStatusWithoutException();
+        StatusOrException statusOrException = new StatusOrException(status);
+        assertNotNull(statusOrException.getStatus());
+        assertSame(status, statusOrException.getStatus());
+        assertNull(statusOrException.getException());
+    }
+
+    /**
+     * Verifies that a {@link StatusOrException} constructed with an {@link Exception} exposes it via
+     * {@link StatusOrException#getException()} and returns null from {@link StatusOrException#getStatus()}.
+     */
+    public void testStatusOrExceptionWithException() {
+        String message = randomAlphaOfLengthBetween(5, 20);
+        Exception exception = new ElasticsearchException(message);
+        StatusOrException statusOrException = new StatusOrException(exception);
+        assertNull(statusOrException.getStatus());
+        assertNotNull(statusOrException.getException());
+        assertSame(exception, statusOrException.getException());
+        assertThat(statusOrException.getException().getMessage(), containsString(message));
+    }
+
+    /**
+     * Verifies that {@link StatusOrException#equals(Object)} returns false for null and for an object of a different class.
+     */
+    public void testEqualsReturnsFalseForNullAndWrongType() {
+        StatusOrException statusOrException = createTestInstanceWithoutExceptions();
+        assertFalse(statusOrException.equals(null));
+        assertFalse(statusOrException.equals(BulkByScrollTaskStatusTests.randomStatus()));
+    }
+
+    /**
+     * Verifies that two {@link StatusOrException} instances with the same status are equal and have the same hashCode.
+     */
+    public void testEqualsAndHashCodeWhenHoldingSameStatus() {
+        BulkByScrollTask.Status status = BulkByScrollTaskStatusTests.randomStatusWithoutException();
+        StatusOrException first = new StatusOrException(status);
+        StatusOrException second = new StatusOrException(status);
+        assertEquals(first, second);
+        assertEquals(first.hashCode(), second.hashCode());
+    }
+
     /**
      * Test parsing {@link StatusOrException} with inner failures as they don't support asserting on xcontent equivalence, given that
      * exceptions are not parsed back as the same original class. We run the usual {@link AbstractXContentTestCase#testFromXContent()}

server/src/test/java/org/elasticsearch/index/reindex/BulkByScrollTaskStatusOrExceptionWireSerializingTests.java

@@ -0,0 +1,141 @@
+/*
+ * 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.ElasticsearchException;
+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.Objects;
+
+/**
+ * Wire serialization tests for {@link BulkByScrollTask.StatusOrException}.
+ * Uses a wrapper type so that equality after round-trip is semantic (e.g. exception
+ * messages instead of instance identity), matching
+ * {@link BulkByScrollTaskStatusOrExceptionTests#assertEqualStatusOrException}.
+ */
+public class BulkByScrollTaskStatusOrExceptionWireSerializingTests extends AbstractWireSerializingTestCase<
+    BulkByScrollTaskStatusOrExceptionWireSerializingTests.StatusOrExceptionWrapper> {
+
+    @Override
+    protected Writeable.Reader<StatusOrExceptionWrapper> instanceReader() {
+        return StatusOrExceptionWrapper::new;
+    }
+
+    @Override
+    protected StatusOrExceptionWrapper createTestInstance() {
+        BulkByScrollTask.StatusOrException statusOrException = BulkByScrollTaskStatusOrExceptionTests.createTestInstanceWithExceptions();
+        return new StatusOrExceptionWrapper(statusOrException);
+    }
+
+    @Override
+    protected StatusOrExceptionWrapper mutateInstance(StatusOrExceptionWrapper instance) throws IOException {
+        BulkByScrollTask.StatusOrException statusOrException = instance.statusOrException;
+        int field = between(0, 1);
+        if (field == 0) {
+            if (statusOrException.getStatus() != null) {
+                return new StatusOrExceptionWrapper(
+                    new BulkByScrollTask.StatusOrException(
+                        BulkByScrollTaskStatusWireSerializingTests.mutateStatus(statusOrException.getStatus())
+                    )
+                );
+            } else {
+                return new StatusOrExceptionWrapper(new BulkByScrollTask.StatusOrException(BulkByScrollTaskStatusTests.randomStatus()));
+            }
+        } else {
+            if (statusOrException.getException() != null) {
+                Exception currentException = statusOrException.getException();
+                Exception differentException = randomValueOtherThan(
+                    currentException,
+                    () -> new ElasticsearchException(randomAlphaOfLengthBetween(5, 15))
+                );
+                return new StatusOrExceptionWrapper(new BulkByScrollTask.StatusOrException(differentException));
+            } else {
+                return new StatusOrExceptionWrapper(
+                    new BulkByScrollTask.StatusOrException(new ElasticsearchException(randomAlphaOfLengthBetween(5, 15)))
+                );
+            }
+        }
+    }
+
+    @Override
+    protected void assertEqualInstances(StatusOrExceptionWrapper expectedInstance, StatusOrExceptionWrapper newInstance) {
+        assertNotSame(expectedInstance, newInstance);
+        BulkByScrollTaskStatusOrExceptionTests.assertEqualStatusOrException(
+            expectedInstance.statusOrException,
+            newInstance.statusOrException,
+            true,
+            true
+        );
+    }
+
+    /**
+     * Wrapper around {@link BulkByScrollTask.StatusOrException} that implements semantic equality and hashCode
+     * so that round-trip serialization passes (e.g. exceptions are compared by message, not reference).
+     */
+    static class StatusOrExceptionWrapper implements Writeable {
+        private final BulkByScrollTask.StatusOrException statusOrException;
+
+        StatusOrExceptionWrapper(BulkByScrollTask.StatusOrException statusOrException) {
+            this.statusOrException = statusOrException;
+        }
+
+        StatusOrExceptionWrapper(StreamInput in) throws IOException {
+            this.statusOrException = new BulkByScrollTask.StatusOrException(in);
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            statusOrException.writeTo(out);
+        }
+
+        @Override
+        public boolean equals(Object other) {
+            if (this == other) return true;
+            if (other == null || getClass() != other.getClass()) return false;
+            StatusOrExceptionWrapper that = (StatusOrExceptionWrapper) other;
+            return statusOrExceptionEquals(statusOrException, that.statusOrException);
+        }
+
+        @Override
+        public int hashCode() {
+            return statusOrExceptionHashCode(statusOrException);
+        }
+
+        private static boolean statusOrExceptionEquals(
+            BulkByScrollTask.StatusOrException first,
+            BulkByScrollTask.StatusOrException second
+        ) {
+            if (first == second) return true;
+            if (first == null || second == null) return false;
+            if (first.getStatus() != null && second.getStatus() != null) {
+                return BulkByScrollTaskStatusWireSerializingTests.StatusWrapper.statusEquals(first.getStatus(), second.getStatus());
+            }
+            if (first.getException() != null && second.getException() != null) {
+                return Objects.equals(first.getException().getMessage(), second.getException().getMessage());
+            }
+            return false;
+        }
+
+        private static int statusOrExceptionHashCode(BulkByScrollTask.StatusOrException statusOrException) {
+            if (statusOrException == null) return 0;
+            if (statusOrException.getStatus() != null) {
+                return BulkByScrollTaskStatusWireSerializingTests.StatusWrapper.statusHashCode(statusOrException.getStatus());
+            }
+            if (statusOrException.getException() != null) {
+                return Objects.hashCode(statusOrException.getException().getMessage());
+            }
+            return 0;
+        }
+    }
+}

server/src/test/java/org/elasticsearch/index/reindex/BulkByScrollTaskStatusTests.java

@@ -118,7 +118,7 @@ public static BulkByScrollTask.Status randomStatusWithoutException() {
         return new BulkByScrollTask.Status(statuses, randomBoolean() ? "test" : null);
     }
 
-    private static BulkByScrollTask.Status randomWorkingStatus(Integer sliceId) {
+    public static BulkByScrollTask.Status randomWorkingStatus(Integer sliceId) {
         // These all should be believably small because we sum them if we have multiple workers
         int total = between(0, 10000000);
         int updated = between(0, total);
@@ -369,4 +369,66 @@ protected ToXContent.Params getToXContentParams() {
         }
         return new ToXContent.MapParams(params);
     }
+
+    /**
+     * Verifies that {@link BulkByScrollTask.Status#getWriteableName()} returns the expected name used for wire serialization.
+     */
+    public void testGetWriteableName() {
+        BulkByScrollTask.Status status = randomStatusWithoutException();
+        assertEquals(BulkByScrollTask.Status.NAME, status.getWriteableName());
+    }
+
+    /**
+     * Verifies that {@link BulkByScrollTask.Status#equalsWithoutSliceStatus(Object, boolean, boolean)} treats two statuses
+     * that differ only in {@code updated} as equal when {@code includeUpdated} is false.
+     */
+    public void testEqualsWithoutSliceStatusRespectsIncludeUpdated() {
+        BulkByScrollTask.Status status = randomWorkingStatus(null);
+        long otherUpdated = randomValueOtherThan(status.getUpdated(), () -> (long) between(0, 10000));
+        BulkByScrollTask.Status sameExceptUpdated = new BulkByScrollTask.Status(
+            status.getSliceId(),
+            status.getTotal(),
+            otherUpdated,
+            status.getCreated(),
+            status.getDeleted(),
+            status.getBatches(),
+            status.getVersionConflicts(),
+            status.getNoops(),
+            status.getBulkRetries(),
+            status.getSearchRetries(),
+            status.getThrottled(),
+            status.getRequestsPerSecond(),
+            status.getReasonCancelled(),
+            status.getThrottledUntil()
+        );
+        assertTrue(status.equalsWithoutSliceStatus(sameExceptUpdated, false, true));
+        assertFalse(status.equalsWithoutSliceStatus(sameExceptUpdated, true, true));
+    }
+
+    /**
+     * Verifies that {@link BulkByScrollTask.Status#equalsWithoutSliceStatus(Object, boolean, boolean)} treats two statuses
+     * that differ only in {@code created} as equal when {@code includeCreated} is false.
+     */
+    public void testEqualsWithoutSliceStatusRespectsIncludeCreated() {
+        BulkByScrollTask.Status status = randomWorkingStatus(null);
+        long otherCreated = randomValueOtherThan(status.getCreated(), () -> (long) between(0, 10000));
+        BulkByScrollTask.Status sameExceptCreated = new BulkByScrollTask.Status(
+            status.getSliceId(),
+            status.getTotal(),
+            status.getUpdated(),
+            otherCreated,
+            status.getDeleted(),
+            status.getBatches(),
+            status.getVersionConflicts(),
+            status.getNoops(),
+            status.getBulkRetries(),
+            status.getSearchRetries(),
+            status.getThrottled(),
+            status.getRequestsPerSecond(),
+            status.getReasonCancelled(),
+            status.getThrottledUntil()
+        );
+        assertTrue(status.equalsWithoutSliceStatus(sameExceptCreated, true, false));
+        assertFalse(status.equalsWithoutSliceStatus(sameExceptCreated, true, true));
+    }
 }