[FLINK-15797][k8s] Reduce log noise of Fabric8FlinkKubeClient

This commit reduces the log noise of the Fabric8FlinkKubeClient by changing the log level
from info to debug.

This closes apache#10965.

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java

@@ -123,7 +123,7 @@ public void createConfigMap() {
  configMap = c.decorate(configMap);
  }
 
- LOG.info("Create config map with data size {}", configMap.getInternalResource().getData().size());
+ LOG.debug("Create config map with data size {}", configMap.getInternalResource().getData().size());
  this.internalClient.configMaps().create(configMap.getInternalResource());
  }
 
@@ -146,7 +146,8 @@ public void createFlinkMasterDeployment(ClusterSpecification clusterSpecificatio
  }
 
  deployment = new FlinkMasterDeploymentDecorator(clusterSpecification).decorate(deployment);
- LOG.info("Create Flink Master deployment with spec: {}", deployment.getInternalResource().getSpec());
+
+ LOG.debug("Create Flink Master deployment with spec: {}", deployment.getInternalResource().getSpec());
 
  this.internalClient
  .apps()
@@ -164,7 +165,8 @@ public void createTaskManagerPod(TaskManagerPodParameter parameter) {
  }
 
  pod = new TaskManagerPodDecorator(parameter).decorate(pod);
- LOG.info("Create TaskManager pod with spec: {}", pod.getInternalResource().getSpec().toString());
+
+ LOG.debug("Create TaskManager pod with spec: {}", pod.getInternalResource().getSpec());
 
  this.internalClient.pods().inNamespace(this.nameSpace).create(pod.getInternalResource());
  }
@@ -234,7 +236,7 @@ public void stopAndCleanupCluster(String clusterId) {
 
  @Override
  public void handleException(Exception e) {
- LOG.error("Encounter Kubernetes Exception.", e);
+ LOG.error("A Kubernetes exception occurred.", e);
  }
 
  @Override
@@ -254,7 +256,7 @@ public void watchPodsAndDoCallback(Map<String, String> labels, PodCallbackHandle
  final Watcher<Pod> watcher = new Watcher<Pod>() {
  @Override
  public void eventReceived(Action action, Pod pod) {
- LOG.info("Received {} event for pod {}, details: {}", action, pod.getMetadata().getName(), pod.getStatus());
+ LOG.debug("Received {} event for pod {}, details: {}", action, pod.getMetadata().getName(), pod.getStatus());
  switch (action) {
  case ADDED:
  callbackHandler.onAdded(Collections.singletonList(new KubernetesPod(flinkConfig, pod)));
@@ -269,14 +271,14 @@ public void eventReceived(Action action, Pod pod) {
  callbackHandler.onDeleted(Collections.singletonList(new KubernetesPod(flinkConfig, pod)));
  break;
  default:
- LOG.info("Skip handling {} event for pod {}", action, pod.getMetadata().getName());
+ LOG.debug("Ignore handling {} event for pod {}", action, pod.getMetadata().getName());
  break;
  }
  }
 
  @Override
  public void onClose(KubernetesClientException e) {
- LOG.error("Pods watcher onClose", e);
+ LOG.error("The pods watcher is closing.", e);
  }
  };
  this.internalClient.pods().withLabels(labels).watch(watcher);
@@ -295,7 +297,7 @@ private CompletableFuture<KubernetesService> createService(
  kubernetesService = d.decorate(kubernetesService);
  }
 
- LOG.info("Create service with spec: {}", kubernetesService.getInternalResource().getSpec());
+ LOG.debug("Create service {} with spec: {}", serviceName, kubernetesService.getInternalResource().getSpec());
 
  this.internalClient.services().create(kubernetesService.getInternalResource());