From 5954354e6200f8fa1d3304e5bff304adf2b17aa8 Mon Sep 17 00:00:00 2001
From: Jason Tedor <jason@tedor.me>
Date: Tue, 28 Aug 2018 08:53:45 -0400
Subject: [PATCH] Fix ShardFollowNodeTask.Status equals and hash code (#33189)

These were broken when fetch exceptions were introduced to the status
object but equals and hash code were not updated then. This commit
addresses that.
---
 .../xpack/ccr/action/ShardFollowNodeTask.java | 24 ++++++++++++++++---
 1 file changed, 21 insertions(+), 3 deletions(-)

diff --git a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowNodeTask.java b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowNodeTask.java
index 6854a9f5741..80d6ed4cb4a 100644
--- a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowNodeTask.java
+++ b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowNodeTask.java
@@ -687,7 +687,7 @@ public abstract class ShardFollowNodeTask extends AllocatedPersistentTask {
             this.numberOfSuccessfulBulkOperations = numberOfSuccessfulBulkOperations;
             this.numberOfFailedBulkOperations = numberOfFailedBulkOperations;
             this.numberOfOperationsIndexed = numberOfOperationsIndexed;
-            this.fetchExceptions = fetchExceptions;
+            this.fetchExceptions = Objects.requireNonNull(fetchExceptions);
         }
 
         public Status(final StreamInput in) throws IOException {
@@ -821,7 +821,15 @@ public abstract class ShardFollowNodeTask extends AllocatedPersistentTask {
                     operationsReceived == that.operationsReceived &&
                     totalTransferredBytes == that.totalTransferredBytes &&
                     numberOfSuccessfulBulkOperations == that.numberOfSuccessfulBulkOperations &&
-                    numberOfFailedBulkOperations == that.numberOfFailedBulkOperations;
+                    numberOfFailedBulkOperations == that.numberOfFailedBulkOperations &&
+                    numberOfOperationsIndexed == that.numberOfOperationsIndexed &&
+                    /*
+                     * ElasticsearchException does not implement equals so we will assume the fetch exceptions are equal if they are equal
+                     * up to the key set and their messages.  Note that we are relying on the fact that the fetch exceptions are ordered by
+                     * keys.
+                     */
+                    fetchExceptions.keySet().equals(that.fetchExceptions.keySet()) &&
+                    getFetchExceptionMessages(this).equals(getFetchExceptionMessages(that));
         }
 
         @Override
@@ -843,8 +851,18 @@ public abstract class ShardFollowNodeTask extends AllocatedPersistentTask {
                     operationsReceived,
                     totalTransferredBytes,
                     numberOfSuccessfulBulkOperations,
-                    numberOfFailedBulkOperations);
+                    numberOfFailedBulkOperations,
+                    numberOfOperationsIndexed,
+                    /*
+                     * ElasticsearchException does not implement hash code so we will compute the hash code based on the key set and the
+                     * messages. Note that we are relying on the fact that the fetch exceptions are ordered by keys.
+                     */
+                    fetchExceptions.keySet(),
+                    getFetchExceptionMessages(this));
+        }
 
+        private static List<String> getFetchExceptionMessages(final Status status) {
+            return status.fetchExceptions().values().stream().map(ElasticsearchException::getMessage).collect(Collectors.toList());
         }
 
         public String toString() {