1996fanrui commented on code in PR #28652:
URL: https://github.com/apache/flink/pull/28652#discussion_r3555410806
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java:
##########
@@ -713,30 +889,110 @@ private void checkAnnouncedOnlyOnce(SequenceBuffer
sequenceBuffer) {
}
/**
- * Spills all queued buffers on checkpoint start. If barrier has already
been received (and
- * reordered), spill only the overtaken buffers.
+ * Persists inflight data on checkpoint start. During recovery, persists
recovered buffers
+ * before the matching RecoveryCheckpointBarrier sentinel; after recovery,
uses the normal
+ * remote-channel barrier sequence tracking and persists overtaken live
buffers.
*/
public void checkpointStarted(CheckpointBarrier barrier) throws
CheckpointException {
- synchronized (receivedBuffers) {
- if (barrier.getId() < lastBarrierId) {
- throw new CheckpointException(
- String.format(
- "Sequence number for checkpoint %d is not
known (it was likely been overwritten by a newer checkpoint %d)",
- barrier.getId(), lastBarrierId),
- CheckpointFailureReason
- .CHECKPOINT_SUBSUMED); // currently, at most
one active unaligned
- // checkpoint is possible
- } else if (barrier.getId() > lastBarrierId) {
- // This channel has received some obsolete barrier, older
compared to the
- // checkpointId
- // which we are processing right now, and we should ignore
that obsoleted checkpoint
- // barrier sequence number.
- resetLastBarrier();
+ try {
+ List<Buffer> toPersist;
+ synchronized (receivedBuffers) {
+ if (inRecovery) {
+ toPersist = collectPreRecoveryBarrier(barrier.getId());
+ } else {
+ if (barrier.getId() < lastBarrierId) {
+ // Currently, at most one active unaligned checkpoint
is possible.
+ throw new CheckpointException(
+ String.format(
+ "Sequence number for checkpoint %d is
not known (it was likely been overwritten by a newer checkpoint %d)",
+ barrier.getId(), lastBarrierId),
+ CheckpointFailureReason.CHECKPOINT_SUBSUMED);
+ } else if (barrier.getId() > lastBarrierId) {
+ // This channel has received some obsolete barrier,
older compared to the
+ // checkpointId which we are processing right now, and
we should ignore that
+ // obsoleted checkpoint barrier sequence number.
+ resetLastBarrier();
+ }
+ toPersist = getInflightBuffersUnsafe(barrier.getId());
+ }
+ channelStatePersister.startPersisting(barrier.getId(),
toPersist);
+ }
+ } catch (IOException e) {
+ throw new CheckpointException(
+ "Failed to extract recovered buffers for checkpoint " +
barrier.getId(),
+ CheckpointFailureReason.CHECKPOINT_DECLINED,
+ e);
+ }
+ }
+
+ /**
+ * Walks {@link #receivedBuffers} (skipping priority events) up to the
{@link
+ * RecoveryCheckpointBarrier} sentinel matching {@code checkpointId},
retaining each pre-barrier
+ * recovered data buffer and removing the sentinel. During recovery the
upstream has no credit,
+ * so {@code receivedBuffers} holds only recovered buffers, sentinels, and
priority events — no
+ * live data buffers.
+ *
+ * @throws IOException if no sentinel matching {@code checkpointId} is
found (the snapshot
+ * protocol guarantees one must be present while the channel is in
recovery).
+ */
+ @GuardedBy("receivedBuffers")
+ private List<Buffer> collectPreRecoveryBarrier(long checkpointId) throws
IOException {
+ assert Thread.holdsLock(receivedBuffers);
+ List<Buffer> retained = new ArrayList<>();
+ SequenceBuffer sentinel = null;
+ try {
+ Iterator<SequenceBuffer> it = receivedBuffers.iterator();
+ // Priority events are stored separately at the head and never
carry recovered data.
+ Iterators.advance(it, receivedBuffers.getNumPriorityElements());
+ while (it.hasNext()) {
+ SequenceBuffer sb = it.next();
+ if (isRecoveryCheckpointBarrier(sb.buffer, checkpointId)) {
+ sentinel = sb;
+ break;
+ }
+ // Skip non-data events (e.g. the
EndOfFetchedChannelStateEvent sentinel appended
+ // after the recovered buffers): only recovered data buffers
are snapshotted.
+ if (sb.buffer.isBuffer()) {
Review Comment:
yes, you are correct.
`RecoveryCheckpointBarrier` is added before step6 since channel is
`inRecovery`.
Added a comment via 6684c478dca58af11c86c088f0a6f4e60aa36c89
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
