[hotfix] Do not use ExecutorService.submit since it can swallow exceptions

This commit changes the KubernetesLeaderElector to use ExecutorService.execute instead of submit
which ensures that potential exceptions are forwarded to the fatal uncaught exeception handler.

This closes apache#15740.

Author: tillrohrmann

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElector.java

@@ -53,6 +53,8 @@ public class KubernetesLeaderElector {
     @VisibleForTesting
     public static final String LEADER_ANNOTATION_KEY = "control-plane.alpha.kubernetes.io/leader";
 
+    private final Object lock = new Object();
+
     private final ExecutorService executorService =
             Executors.newSingleThreadExecutor(
                     new ExecutorThreadFactory("KubernetesLeaderElector-ExecutorService"));
@@ -92,11 +94,20 @@ public KubernetesLeaderElector(
     }
 
     public void run() {
-        executorService.submit(internalLeaderElector::run);
+        synchronized (lock) {
+            if (executorService.isShutdown()) {
+                LOG.debug(
+                        "Ignoring KubernetesLeaderElector.run call because the leader elector has already been shut down.");
+            } else {
+                executorService.execute(internalLeaderElector::run);
+            }
+        }
     }
 
     public void stop() {
-        executorService.shutdownNow();
+        synchronized (lock) {
+            executorService.shutdownNow();
+        }
     }
 
     public static boolean hasLeadership(KubernetesConfigMap configMap, String lockIdentity) {