no break + event

Author: zstan

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java

@@ -1354,7 +1354,6 @@ private void processDhtTxSalvageRequest(GridDhtTxSalvageMessage req) {
                 // GridDhtTxLocal possible
                 if (active instanceof GridDhtTxRemote) {
                     ctx.tm().salvageTx(active);
-                    break;
                 }
             }
         }

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java

@@ -3140,6 +3140,7 @@ private TxRecoveryInitRunnable(ClusterNode node) {
 
                 for (final IgniteInternalTx tx : activeTransactions()) {
                     Map<UUID, Collection<UUID>> txNodes = tx.transactionNodes();
+                    boolean localInMaster = tx.masterNodeIds().contains(cctx.localNodeId());
 
                     if (tx.storeWriteThrough() && txNodes != null
                         && tx.near() && txNodes.containsKey(evtNodeId)
@@ -3149,7 +3150,11 @@ private TxRecoveryInitRunnable(ClusterNode node) {
                         sendTxSalvage(tx, evtNodeId);
                     }
 
-                    if (tx.storeWriteThrough() && !tx.masterNodeIds().contains(cctx.localNodeId())
+                    if (tx.storeWriteThrough() && tx.masterNodeIds().contains(evtNodeId)) {
+                        if (localInMaster || tx.eventNodeId().equals(evtNodeId))
+                            salvageTx(tx, RECOVERY_FINISH);
+                    }
+                    else if (tx.storeWriteThrough() && !localInMaster
                         && tx.nodeId().equals(evtNodeId) && tx.state() == PREPARED) {
                         boolean fullSyncedOp = tx.writeEntries().stream().map(e ->
                             cctx.cacheContext(e.cacheId())).allMatch(GridCacheContext::syncCommit);