From 7fcffebe125708d9060b29f0cd5fb31f4efc6acb Mon Sep 17 00:00:00 2001 From: Purushottam Sinha Date: Fri, 5 Jun 2026 14:02:21 +0530 Subject: [PATCH] [FLINK-39860][runtime] Reduce ArchUnit violations: production code must not call @VisibleForTesting methods All 23 frozen violations of the "Production code must not call methods annotated with @VisibleForTesting" rule are resolved. In every case the annotation was inaccurate: the target method is genuinely called by production code, so the test-only marker misled maintainers. Removed the annotation from the 12 affected methods (no visibility or signature change) and dropped two now-unused imports. The rule's freeze store is emptied; the freeze() wrapper is kept so the rule now enforces zero such violations going forward. --- .../e5126cae-f3fe-48aa-b6fb-60ae6cc3fcd5 | 23 ------------------- .../flink/cep/nfa/sharedbuffer/Lockable.java | 2 -- .../apache/flink/runtime/blob/BlobKey.java | 1 - .../executiongraph/ExecutionVertex.java | 1 - .../nonha/embedded/EmbeddedLeaderService.java | 2 -- .../io/network/partition/DataSetMetaInfo.java | 2 -- .../RecreateOnResetOperatorCoordinator.java | 1 - .../state/heap/CopyOnWriteStateMap.java | 1 - .../TaskExecutorResourceUtils.java | 2 -- .../api/operators/SourceOperator.java | 5 ++-- .../tasks/mailbox/MailboxProcessor.java | 1 - 11 files changed, 2 insertions(+), 39 deletions(-) diff --git a/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/e5126cae-f3fe-48aa-b6fb-60ae6cc3fcd5 b/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/e5126cae-f3fe-48aa-b6fb-60ae6cc3fcd5 index f5af66381c015..e69de29bb2d1d 100644 --- a/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/e5126cae-f3fe-48aa-b6fb-60ae6cc3fcd5 +++ b/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/e5126cae-f3fe-48aa-b6fb-60ae6cc3fcd5 @@ -1,23 +0,0 @@ -Constructor (org.apache.flink.runtime.state.heap.CopyOnWriteStateMap)> calls method in (CopyOnWriteStateMapSnapshot.java:86) -Constructor (org.apache.flink.streaming.api.operators.SourceOperator, java.util.function.Consumer, int, int)> calls method in (StreamTaskExternallyInducedSourceInput.java:41) -Method calls method in (LockableTypeSerializerSnapshot.java:58) -Method calls method in (BlobInputStream.java:127) -Method calls method in (BlobInputStream.java:163) -Method calls method in (BlobOutputStream.java:186) -Method calls method in (BlobUtils.java:665) -Method calls method in (FileSystemBlobStore.java:148) -Method calls method in (Execution.java:1202) -Method calls method in (EmbeddedHaServicesWithLeadershipControl.java:83) -Method calls method in (EmbeddedHaServicesWithLeadershipControl.java:95) -Method calls method in (EmbeddedHaServicesWithLeadershipControl.java:109) -Method calls method in (EmbeddedHaServicesWithLeadershipControl.java:77) -Method calls method in (EmbeddedHaServicesWithLeadershipControl.java:89) -Method calls method in (EmbeddedHaServicesWithLeadershipControl.java:102) -Method calls method in (ResourceManagerPartitionTrackerImpl.java:286) -Method calls method in (RecreateOnResetOperatorCoordinator.java:383) -Method calls method in (TaskManagerConfiguration.java:238) -Method calls method in (TaskManagerConfiguration.java:240) -Method calls method in (TaskManagerServices.java:481) -Method calls method in (TaskManagerServices.java:479) -Method calls method in (SourceOperatorStreamTask.java:101) -Method calls method in (MailboxExecutorImpl.java:64) diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/Lockable.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/Lockable.java index f80539125773a..ab22e6d11bc16 100644 --- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/Lockable.java +++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/Lockable.java @@ -18,7 +18,6 @@ package org.apache.flink.cep.nfa.sharedbuffer; -import org.apache.flink.annotation.VisibleForTesting; import org.apache.flink.api.common.typeutils.CompositeTypeSerializerUtil; import org.apache.flink.api.common.typeutils.LegacySerializerSnapshotTransformer; import org.apache.flink.api.common.typeutils.TypeSerializer; @@ -186,7 +185,6 @@ public TypeSerializerSnapshot> snapshotConfiguration() { return new LockableTypeSerializerSnapshot<>(this); } - @VisibleForTesting TypeSerializer getElementSerializer() { return elementSerializer; } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobKey.java b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobKey.java index bf89636df1c42..8b74b68166c38 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobKey.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobKey.java @@ -161,7 +161,6 @@ static BlobKey createKey(BlobType type, byte[] key, byte[] random) { * * @return a 20 bit hash of the contents the key refers to */ - @VisibleForTesting public byte[] getHash() { return key; } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java index 81d70ec620280..ad879e5d2af24 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java @@ -502,7 +502,6 @@ void cachePartitionInfo(PartitionInfo partitionInfo) { * * @return list of finished partitions. */ - @VisibleForTesting public List finishPartitionsIfNeeded() { List finishedPartitions = null; MarkPartitionFinishedStrategy markPartitionFinishedStrategy = diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedLeaderService.java b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedLeaderService.java index 68e08f4e36cc0..c18be31bc51e3 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedLeaderService.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedLeaderService.java @@ -382,7 +382,6 @@ private void removeListener(EmbeddedLeaderRetrievalService service) { } } - @VisibleForTesting CompletableFuture grantLeadership() { synchronized (lock) { if (shutdown) { @@ -398,7 +397,6 @@ private CompletableFuture getShutDownFuture() { new FlinkException("EmbeddedLeaderService has been shut down.")); } - @VisibleForTesting CompletableFuture revokeLeadership() { synchronized (lock) { if (shutdown) { diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/DataSetMetaInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/DataSetMetaInfo.java index cd7bb7a0c0bb6..a933cb43bf126 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/DataSetMetaInfo.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/DataSetMetaInfo.java @@ -17,7 +17,6 @@ package org.apache.flink.runtime.io.network.partition; -import org.apache.flink.annotation.VisibleForTesting; import org.apache.flink.runtime.shuffle.ShuffleDescriptor; import org.apache.flink.util.Preconditions; @@ -71,7 +70,6 @@ static DataSetMetaInfo withoutNumRegisteredPartitions(int numTotalPartitions) { return new DataSetMetaInfo(UNKNOWN, numTotalPartitions); } - @VisibleForTesting public static DataSetMetaInfo withNumRegisteredPartitions( int numRegisteredPartitions, int numTotalPartitions) { Preconditions.checkArgument(numRegisteredPartitions > 0); diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/RecreateOnResetOperatorCoordinator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/RecreateOnResetOperatorCoordinator.java index 98718fe3ba839..6fd38ad0f71a3 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/RecreateOnResetOperatorCoordinator.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/RecreateOnResetOperatorCoordinator.java @@ -293,7 +293,6 @@ public CheckpointCoordinator getCheckpointCoordinator() { return context.getCheckpointCoordinator(); } - @VisibleForTesting synchronized void quiesce() { quiesced = true; } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/CopyOnWriteStateMap.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/CopyOnWriteStateMap.java index 6ad7401be81b5..dae3b7d54bbdf 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/CopyOnWriteStateMap.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/CopyOnWriteStateMap.java @@ -471,7 +471,6 @@ void releaseSnapshot(int snapshotVersion) { * Creates (combined) copy of the table arrays for a snapshot. This method must be called by the * same Thread that does modifications to the {@link CopyOnWriteStateMap}. */ - @VisibleForTesting @SuppressWarnings("unchecked") StateMapEntry[] snapshotMapArrays() { diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorResourceUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorResourceUtils.java index 1461afb60120a..90fad05d2279b 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorResourceUtils.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorResourceUtils.java @@ -112,7 +112,6 @@ private static void checkTaskExecutorNetworkConfigSet(ReadableConfig config) { * This must be consist with {@link * org.apache.flink.runtime.resourcemanager.slotmanager.SlotManagerUtils#generateDefaultSlotResourceProfile}. */ - @VisibleForTesting public static ResourceProfile generateDefaultSlotResourceProfile( TaskExecutorResourceSpec taskExecutorResourceSpec, int numberOfSlots) { final ResourceProfile.Builder resourceProfileBuilder = @@ -137,7 +136,6 @@ public static ResourceProfile generateDefaultSlotResourceProfile( return resourceProfileBuilder.build(); } - @VisibleForTesting public static ResourceProfile generateTotalAvailableResourceProfile( TaskExecutorResourceSpec taskExecutorResourceSpec) { return ResourceProfile.newBuilder() diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java index f153617b2c4fe..67be8dd65ecfe 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java @@ -928,13 +928,12 @@ private void registerReader(List splits) throws Exception { splitSerializer)); } - // --------------- methods for unit tests ------------ - - @VisibleForTesting public SourceReader getSourceReader() { return sourceReader; } + // --------------- methods for unit tests ------------ + @VisibleForTesting ListState getReaderState() { return readerState; diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java index 4e976eca64422..e8f732929c73a 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java @@ -455,7 +455,6 @@ private MailboxDefaultAction.Suspension suspendDefaultAction( return suspendedDefaultAction; } - @VisibleForTesting public boolean isDefaultActionAvailable() { return suspendedDefaultAction == null; }