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.

Author: jasontedor

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowNodeTask.java

@@ -687,7 +687,7 @@ public NavigableMap<Long, ElasticsearchException> fetchExceptions() {
             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 boolean equals(final Object o) {
                     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 int hashCode() {
                     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() {