[FLINK-39860][runtime] Reduce ArchUnit violations: production code must not call @VisibleForTesting methods

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.CopyOnWriteStateMapSnapshot.<init>(org.apache.flink.runtime.state.heap.CopyOnWriteStateMap)> calls method <org.apache.flink.runtime.state.heap.CopyOnWriteStateMap.snapshotMapArrays()> in (CopyOnWriteStateMapSnapshot.java:86)
-Constructor <org.apache.flink.streaming.runtime.io.StreamTaskExternallyInducedSourceInput.<init>(org.apache.flink.streaming.api.operators.SourceOperator, java.util.function.Consumer, int, int)> calls method <org.apache.flink.streaming.api.operators.SourceOperator.getSourceReader()> in (StreamTaskExternallyInducedSourceInput.java:41)
-Method <org.apache.flink.cep.nfa.sharedbuffer.LockableTypeSerializerSnapshot.getNestedSerializers(org.apache.flink.cep.nfa.sharedbuffer.Lockable$LockableTypeSerializer)> calls method <org.apache.flink.cep.nfa.sharedbuffer.Lockable$LockableTypeSerializer.getElementSerializer()> in (LockableTypeSerializerSnapshot.java:58)
-Method <org.apache.flink.runtime.blob.BlobInputStream.read()> calls method <org.apache.flink.runtime.blob.BlobKey.getHash()> in (BlobInputStream.java:127)
-Method <org.apache.flink.runtime.blob.BlobInputStream.read([B, int, int)> calls method <org.apache.flink.runtime.blob.BlobKey.getHash()> in (BlobInputStream.java:163)
-Method <org.apache.flink.runtime.blob.BlobOutputStream.receiveAndCheckPutResponse(java.io.InputStream, java.security.MessageDigest, org.apache.flink.runtime.blob.BlobKey$BlobType)> calls method <org.apache.flink.runtime.blob.BlobKey.getHash()> in (BlobOutputStream.java:186)
-Method <org.apache.flink.runtime.blob.BlobUtils.checkAndDeleteCorruptedBlobs(java.nio.file.Path, org.slf4j.Logger)> calls method <org.apache.flink.runtime.blob.BlobKey.getHash()> in (BlobUtils.java:665)
-Method <org.apache.flink.runtime.blob.FileSystemBlobStore.get(java.lang.String, java.io.File, org.apache.flink.runtime.blob.BlobKey)> calls method <org.apache.flink.runtime.blob.BlobKey.getHash()> in (FileSystemBlobStore.java:148)
-Method <org.apache.flink.runtime.executiongraph.Execution.finishPartitionsAndUpdateConsumers()> calls method <org.apache.flink.runtime.executiongraph.ExecutionVertex.finishPartitionsIfNeeded()> in (Execution.java:1202)
-Method <org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedHaServicesWithLeadershipControl.grantDispatcherLeadership()> calls method <org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedLeaderService.grantLeadership()> in (EmbeddedHaServicesWithLeadershipControl.java:83)
-Method <org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedHaServicesWithLeadershipControl.grantJobMasterLeadership(org.apache.flink.api.common.JobID)> calls method <org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedLeaderService.grantLeadership()> in (EmbeddedHaServicesWithLeadershipControl.java:95)
-Method <org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedHaServicesWithLeadershipControl.grantResourceManagerLeadership()> calls method <org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedLeaderService.grantLeadership()> in (EmbeddedHaServicesWithLeadershipControl.java:109)
-Method <org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedHaServicesWithLeadershipControl.revokeDispatcherLeadership()> calls method <org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedLeaderService.revokeLeadership()> in (EmbeddedHaServicesWithLeadershipControl.java:77)
-Method <org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedHaServicesWithLeadershipControl.revokeJobMasterLeadership(org.apache.flink.api.common.JobID)> calls method <org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedLeaderService.revokeLeadership()> in (EmbeddedHaServicesWithLeadershipControl.java:89)
-Method <org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedHaServicesWithLeadershipControl.revokeResourceManagerLeadership()> calls method <org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedLeaderService.revokeLeadership()> in (EmbeddedHaServicesWithLeadershipControl.java:102)
-Method <org.apache.flink.runtime.io.network.partition.ResourceManagerPartitionTrackerImpl.listDataSets()> calls method <org.apache.flink.runtime.io.network.partition.DataSetMetaInfo.withNumRegisteredPartitions(int, int)> in (ResourceManagerPartitionTrackerImpl.java:286)
-Method <org.apache.flink.runtime.operators.coordination.RecreateOnResetOperatorCoordinator$DeferrableCoordinator.closeAsync(long)> calls method <org.apache.flink.runtime.operators.coordination.RecreateOnResetOperatorCoordinator$QuiesceableContext.quiesce()> in (RecreateOnResetOperatorCoordinator.java:383)
-Method <org.apache.flink.runtime.taskexecutor.TaskManagerConfiguration.fromConfiguration(org.apache.flink.configuration.Configuration, org.apache.flink.runtime.taskexecutor.TaskExecutorResourceSpec, java.lang.String, java.io.File)> calls method <org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils.generateDefaultSlotResourceProfile(org.apache.flink.runtime.taskexecutor.TaskExecutorResourceSpec, int)> in (TaskManagerConfiguration.java:238)
-Method <org.apache.flink.runtime.taskexecutor.TaskManagerConfiguration.fromConfiguration(org.apache.flink.configuration.Configuration, org.apache.flink.runtime.taskexecutor.TaskExecutorResourceSpec, java.lang.String, java.io.File)> calls method <org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils.generateTotalAvailableResourceProfile(org.apache.flink.runtime.taskexecutor.TaskExecutorResourceSpec)> in (TaskManagerConfiguration.java:240)
-Method <org.apache.flink.runtime.taskexecutor.TaskManagerServices.createTaskSlotTable(int, org.apache.flink.runtime.taskexecutor.TaskExecutorResourceSpec, long, int, java.util.concurrent.Executor)> calls method <org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils.generateDefaultSlotResourceProfile(org.apache.flink.runtime.taskexecutor.TaskExecutorResourceSpec, int)> in (TaskManagerServices.java:481)
-Method <org.apache.flink.runtime.taskexecutor.TaskManagerServices.createTaskSlotTable(int, org.apache.flink.runtime.taskexecutor.TaskExecutorResourceSpec, long, int, java.util.concurrent.Executor)> calls method <org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils.generateTotalAvailableResourceProfile(org.apache.flink.runtime.taskexecutor.TaskExecutorResourceSpec)> in (TaskManagerServices.java:479)
-Method <org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask.init()> calls method <org.apache.flink.streaming.api.operators.SourceOperator.getSourceReader()> in (SourceOperatorStreamTask.java:101)
-Method <org.apache.flink.streaming.runtime.tasks.mailbox.MailboxExecutorImpl.isIdle()> calls method <org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.isDefaultActionAvailable()> in (MailboxExecutorImpl.java:64)

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<Lockable<E>> snapshotConfiguration() {
             return new LockableTypeSerializerSnapshot<>(this);
         }
 
-        @VisibleForTesting
         TypeSerializer<E> getElementSerializer() {
             return elementSerializer;
         }

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;
     }

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<IntermediateResultPartition> finishPartitionsIfNeeded() {
         List<IntermediateResultPartition> finishedPartitions = null;
         MarkPartitionFinishedStrategy markPartitionFinishedStrategy =

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<Void> grantLeadership() {
         synchronized (lock) {
             if (shutdown) {
@@ -398,7 +397,6 @@ private CompletableFuture<Void> getShutDownFuture() {
                 new FlinkException("EmbeddedLeaderService has been shut down."));
     }
 
-    @VisibleForTesting
     CompletableFuture<Void> revokeLeadership() {
         synchronized (lock) {
             if (shutdown) {

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);

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;
         }

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<K, N, S>[] snapshotMapArrays() {
 

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()

flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java

@@ -928,13 +928,12 @@ private void registerReader(List<SplitT> splits) throws Exception {
                         splitSerializer));
     }
 
-    // --------------- methods for unit tests ------------
-
-    @VisibleForTesting
     public SourceReader<OUT, SplitT> getSourceReader() {
         return sourceReader;
     }
 
+    // --------------- methods for unit tests ------------
+
     @VisibleForTesting
     ListState<SplitT> getReaderState() {
         return readerState;

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;
     }