[Java] Reset ClusterBackup state if the Cluster node from which Clust…

…erBackup is replaying the log is not "available", i.e. either is no longer eligible (i.e. after an election) or the backup query cannot be sent to it (e.g. ConsensusModule is down).

This ensures that all the data received by the ClusterBackup is coming from a single source. Previously it was possible to get a replay from one Cluster node while getting snapshots and updating RecordingLog using another node.

aeron-cluster/src/main/java/io/aeron/cluster/ClusterBackupAgent.java

@@ -475,16 +475,12 @@ private void onBackupResponse(
         else if (!logSourceValidator.isAcceptable(leaderMemberId, memberId))
         {
             consensusPublicationGroup.closeAndExcludeCurrent();
-            if (null != logSupplierMember && logSupplierMember.id() == memberId)
-            {
-                // we can no longer replay from the current node due to the role change
-                state(RESET_BACKUP, epochClock.time());
-            }
-            else
-            {
-                // just query another node
-                timeOfLastBackupQueryMs = 0;
-            }
+            state(RESET_BACKUP, epochClock.time());
+            return;
+        }
+        else if (null != logSupplierMember && logSupplierMember.id() != memberId)
+        {
+            state(RESET_BACKUP, epochClock.time());
             return;
         }