(flink) 03/03: [hotfix] Additional logs in AdaptiveScheduler components

[mapohl]

This is an automated email from the ASF dual-hosted git repository.

mapohl pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit 32dc6c019c036b7fea4532bf429d00bfc9b75c78
Author: Zdenek Tison <zti...@confluent.io>
AuthorDate: Fri Sep 20 15:33:50 2024 +0200

    [hotfix] Additional logs in AdaptiveScheduler components
---
 .../scheduler/adaptive/DefaultStateTransitionManager.java        | 2 ++
 .../java/org/apache/flink/runtime/scheduler/adaptive/State.java  | 9 ++++++++-
 2 files changed, 10 insertions(+), 1 deletion(-)

diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManager.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManager.java
index da3ccab27c2..4c7be2ed349 100644
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManager.java
+++ 
b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManager.java
@@ -109,11 +109,13 @@ public class DefaultStateTransitionManager implements 
StateTransitionManager {
 
     @Override
     public void onChange() {
+        LOG.debug("OnChange event received in phase {}.", getPhase());
         phase.onChange();
     }
 
     @Override
     public void onTrigger() {
+        LOG.debug("OnTrigger event received in phase {}.", getPhase());
         phase.onTrigger();
     }
 
diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/State.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/State.java
index 2a4bb966042..7d7de39a439 100644
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/State.java
+++ 
b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/State.java
@@ -103,7 +103,14 @@ interface State extends LabeledGlobalFailureHandler {
             Class<? extends T> clazz, ThrowingConsumer<T, E> action, String 
debugMessage) throws E {
         tryRun(
                 clazz,
-                action,
+                x -> {
+                    getLogger()
+                            .debug(
+                                    "Running '{}' in state {}.",
+                                    debugMessage,
+                                    this.getClass().getSimpleName());
+                    ThrowingConsumer.unchecked(action).accept(x);
+                },
                 logger ->
                         logger.debug(
                                 "Cannot run '{}' because the actual state is 
{} and not {}.",