Call afterWriteOperation after trim translog in peer recovery (#45182)

testShouldFlushAfterPeerRecovery was added #28350 to make sure the
flushing loop triggered by afterWriteOperation eventually terminates.
This test relies on the fact that we call afterWriteOperation after
making changes in translog. In #44756, we roll a new generation in
RecoveryTarget#finalizeRecovery but do not call afterWriteOperation.

Relates #28350
Relates #45073

Author: dnhatn

server/src/main/java/org/elasticsearch/indices/recovery/RecoveryTarget.java

@@ -295,18 +295,20 @@ public void finalizeRecovery(final long globalCheckpoint, final long trimAboveSe
             // Persist the global checkpoint.
             indexShard.sync();
             indexShard.persistRetentionLeases();
-            if (hasUncommittedOperations()) {
-                indexShard.flush(new FlushRequest().force(true).waitIfOngoing(true));
-            }
             if (trimAboveSeqNo != SequenceNumbers.UNASSIGNED_SEQ_NO) {
                 // We should erase all translog operations above trimAboveSeqNo as we have received either the same or a newer copy
                 // from the recovery source in phase2. Rolling a new translog generation is not strictly required here for we won't
                 // trim the current generation. It's merely to satisfy the assumption that the current generation does not have any
                 // operation that would be trimmed (see TranslogWriter#assertNoSeqAbove). This assumption does not hold for peer
                 // recovery because we could have received operations above startingSeqNo from the previous primary terms.
                 indexShard.rollTranslogGeneration();
+                // the flush or translog generation threshold can be reached after we roll a new translog
+                indexShard.afterWriteOperation();
                 indexShard.trimOperationOfPreviousPrimaryTerms(trimAboveSeqNo);
             }
+            if (hasUncommittedOperations()) {
+                indexShard.flush(new FlushRequest().force(true).waitIfOngoing(true));
+            }
             indexShard.finalizeRecovery();
             return null;
         });