You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2023/02/27 13:59:29 UTC
[flink] 01/03: [hotfix] Add missing @GuardedBy annotation for SortMergeResultPartitionReadScheduler and PipelinedSubpartition.
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 5c5cc9dbcc1ce9e041ea3aee9ba32ad6a3e19689
Author: Weijie Guo <re...@163.com>
AuthorDate: Fri Feb 24 23:29:11 2023 +0800
[hotfix] Add missing @GuardedBy annotation for SortMergeResultPartitionReadScheduler and PipelinedSubpartition.
The methods invoked under lock and accessing guarded fields must be annotated as well,
for the inspections/spotbugs check to work properly.
---
.../io/network/partition/PipelinedSubpartition.java | 14 ++++++++++++++
.../partition/SortMergeResultPartitionReadScheduler.java | 9 +++++++++
.../io/network/partition/SortMergeSubpartitionReader.java | 2 ++
3 files changed, 25 insertions(+)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
index 8be8a287836..28dca791ba4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
@@ -215,6 +215,7 @@ public class PipelinedSubpartition extends ResultSubpartition
return newBufferSize;
}
+ @GuardedBy("buffers")
private boolean addBuffer(BufferConsumer bufferConsumer, int partialRecordLength) {
assert Thread.holdsLock(buffers);
if (bufferConsumer.getDataType().hasPriority()) {
@@ -227,6 +228,7 @@ public class PipelinedSubpartition extends ResultSubpartition
return false;
}
+ @GuardedBy("buffers")
private boolean processPriorityBuffer(BufferConsumer bufferConsumer, int partialRecordLength) {
buffers.addPriorityElement(
new BufferConsumerWithPartialRecordLength(bufferConsumer, partialRecordLength));
@@ -261,12 +263,14 @@ public class PipelinedSubpartition extends ResultSubpartition
}
// It just be called after add priorityEvent.
+ @GuardedBy("buffers")
private boolean needNotifyPriorityEvent() {
assert Thread.holdsLock(buffers);
// if subpartition is blocked then downstream doesn't expect any notifications
return buffers.getNumPriorityElements() == 1 && !isBlocked;
}
+ @GuardedBy("buffers")
private void processTimeoutableCheckpointBarrier(BufferConsumer bufferConsumer) {
CheckpointBarrier barrier = parseAndCheckTimeoutableCheckpointBarrier(bufferConsumer);
channelStateWriter.addOutputDataFuture(
@@ -276,6 +280,7 @@ public class PipelinedSubpartition extends ResultSubpartition
createChannelStateFuture(barrier.getId()));
}
+ @GuardedBy("buffers")
private CompletableFuture<List<Buffer>> createChannelStateFuture(long checkpointId) {
assert Thread.holdsLock(buffers);
if (channelStateFuture != null) {
@@ -293,6 +298,7 @@ public class PipelinedSubpartition extends ResultSubpartition
return channelStateFuture;
}
+ @GuardedBy("buffers")
private void completeChannelStateFuture(List<Buffer> channelResult, Throwable e) {
assert Thread.holdsLock(buffers);
if (e != null) {
@@ -303,6 +309,7 @@ public class PipelinedSubpartition extends ResultSubpartition
channelStateFuture = null;
}
+ @GuardedBy("buffers")
private boolean isChannelStateFutureAvailable(long checkpointId) {
assert Thread.holdsLock(buffers);
return channelStateFuture != null && channelStateCheckpointId == checkpointId;
@@ -358,6 +365,7 @@ public class PipelinedSubpartition extends ResultSubpartition
}
}
+ @GuardedBy("buffers")
private boolean findInflightBuffersAndMakeBarrierToPriority(
long checkpointId, List<Buffer> inflightBuffers) throws IOException {
// 1. record the buffers before barrier as inflightBuffers
@@ -544,6 +552,7 @@ public class PipelinedSubpartition extends ResultSubpartition
}
}
+ @GuardedBy("buffers")
private void completeTimeoutableCheckpointBarrier(BufferConsumer bufferConsumer) {
CheckpointBarrier barrier = parseAndCheckTimeoutableCheckpointBarrier(bufferConsumer);
if (!isChannelStateFutureAvailable(barrier.getId())) {
@@ -737,6 +746,7 @@ public class PipelinedSubpartition extends ResultSubpartition
}
/** Gets the number of non-event buffers in this subpartition. */
+ @SuppressWarnings("FieldAccessNotGuarded")
@Override
public int getBuffersInBacklogUnsafe() {
if (isBlocked || buffers.isEmpty()) {
@@ -806,11 +816,15 @@ public class PipelinedSubpartition extends ResultSubpartition
}
/** for testing only. */
+ // suppress this warning as it is only for testing.
+ @SuppressWarnings("FieldAccessNotGuarded")
@VisibleForTesting
CompletableFuture<List<Buffer>> getChannelStateFuture() {
return channelStateFuture;
}
+ // suppress this warning as it is only for testing.
+ @SuppressWarnings("FieldAccessNotGuarded")
@VisibleForTesting
public long getChannelStateCheckpointId() {
return channelStateCheckpointId;
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortMergeResultPartitionReadScheduler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortMergeResultPartitionReadScheduler.java
index 1a5bbae4b54..8330abcf568 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortMergeResultPartitionReadScheduler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortMergeResultPartitionReadScheduler.java
@@ -217,10 +217,14 @@ class SortMergeResultPartitionReadScheduler implements Runnable, BufferRecycler
if (!buffers.isEmpty()) {
return new ArrayDeque<>(buffers);
}
+ // only visibility requirements here.
+ // noinspection FieldAccessNotGuarded
checkState(!isReleased, "Result partition has been already released.");
} while (System.nanoTime() < timeoutTime
|| System.nanoTime() < (timeoutTime = getBufferRequestTimeoutTime()));
+ // only visibility requirements here.
+ // noinspection FieldAccessNotGuarded
if (numRequestedBuffers <= 0) {
throw new TimeoutException(
String.format(
@@ -291,6 +295,7 @@ class SortMergeResultPartitionReadScheduler implements Runnable, BufferRecycler
}
}
+ @GuardedBy("lock")
private void mayNotifyReleased() {
assert Thread.holdsLock(lock);
@@ -361,6 +366,7 @@ class SortMergeResultPartitionReadScheduler implements Runnable, BufferRecycler
}
}
+ @GuardedBy("lock")
private PartitionedFileReader createFileReader(
PartitionedFile resultFile, int targetSubpartition) throws IOException {
assert Thread.holdsLock(lock);
@@ -387,6 +393,7 @@ class SortMergeResultPartitionReadScheduler implements Runnable, BufferRecycler
}
}
+ @GuardedBy("lock")
private void openFileChannels(PartitionedFile resultFile) throws IOException {
assert Thread.holdsLock(lock);
@@ -395,6 +402,7 @@ class SortMergeResultPartitionReadScheduler implements Runnable, BufferRecycler
indexFileChannel = openFileChannel(resultFile.getIndexFilePath());
}
+ @GuardedBy("lock")
private void closeFileChannels() {
assert Thread.holdsLock(lock);
@@ -413,6 +421,7 @@ class SortMergeResultPartitionReadScheduler implements Runnable, BufferRecycler
}
}
+ @GuardedBy("lock")
private void mayTriggerReading() {
assert Thread.holdsLock(lock);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortMergeSubpartitionReader.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortMergeSubpartitionReader.java
index 3cf7a08b77f..9ed821d17ca 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortMergeSubpartitionReader.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortMergeSubpartitionReader.java
@@ -231,6 +231,8 @@ class SortMergeSubpartitionReader
}
}
+ // suppress warning as this method is only for unsafe purpose.
+ @SuppressWarnings("FieldAccessNotGuarded")
@Override
public int unsynchronizedGetNumberOfQueuedBuffers() {
return Math.max(0, buffersRead.size());