gyfora commented on code in PR #1000:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/1000#discussion_r3080393523
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java:
##########
@@ -873,6 +879,110 @@ public JobExceptionsInfoWithHistory getJobExceptions(
}
}
+ @Override
+ public Map<String, String> getJobManagerConfiguration(Configuration conf,
JobID jobId)
+ throws Exception {
+ LOG.debug("Fetching JobManager configuration");
+ try (var clusterClient = getClusterClient(conf)) {
+ var parameters = new JobMessageParameters();
+ parameters.jobPathParameter.resolve(jobId);
+
+ try {
+ var configurationInfo =
+ clusterClient
+ .sendRequest(
+
JobManagerJobConfigurationHeaders.getInstance(),
+ parameters,
+ EmptyRequestBody.getInstance())
+ .get(
+
operatorConfig.getFlinkClientTimeout().toSeconds(),
+ TimeUnit.SECONDS);
+
+ Map<String, String> jmConfig = new HashMap<>();
+ configurationInfo.forEach(entry ->
jmConfig.put(entry.getKey(), entry.getValue()));
+ LOG.debug("Fetched {} JobManager configuration entries",
jmConfig.size());
+ return jmConfig;
+
+ } catch (Exception e) {
+ LOG.warn("Failed to fetch JobManager configuration via REST
API", e);
+ return new HashMap<>();
+ }
Review Comment:
This catch is probably not required here as we already catch outside of this
method, this way we just hide errors unnecessarily
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java:
##########
@@ -873,6 +879,110 @@ public JobExceptionsInfoWithHistory getJobExceptions(
}
}
+ @Override
+ public Map<String, String> getJobManagerConfiguration(Configuration conf,
JobID jobId)
+ throws Exception {
+ LOG.debug("Fetching JobManager configuration");
+ try (var clusterClient = getClusterClient(conf)) {
+ var parameters = new JobMessageParameters();
+ parameters.jobPathParameter.resolve(jobId);
+
+ try {
+ var configurationInfo =
+ clusterClient
+ .sendRequest(
+
JobManagerJobConfigurationHeaders.getInstance(),
+ parameters,
+ EmptyRequestBody.getInstance())
+ .get(
+
operatorConfig.getFlinkClientTimeout().toSeconds(),
+ TimeUnit.SECONDS);
+
+ Map<String, String> jmConfig = new HashMap<>();
+ configurationInfo.forEach(entry ->
jmConfig.put(entry.getKey(), entry.getValue()));
+ LOG.debug("Fetched {} JobManager configuration entries",
jmConfig.size());
+ return jmConfig;
+
+ } catch (Exception e) {
+ LOG.warn("Failed to fetch JobManager configuration via REST
API", e);
+ return new HashMap<>();
+ }
+ }
+ }
+
+ @Override
+ public Map<String, String> getJobConfiguration(Configuration conf, JobID
jobId)
+ throws Exception {
+ LOG.debug("Fetching job configuration");
+ try (var clusterClient = getClusterClient(conf)) {
+ var jobConfigHeaders = JobConfigHeaders.getInstance();
+ var parameters = new JobMessageParameters();
+ parameters.jobPathParameter.resolve(jobId);
+
+ try {
+ JobConfigInfo configurationInfo =
+ clusterClient
+ .sendRequest(
+ jobConfigHeaders,
+ parameters,
+ EmptyRequestBody.getInstance())
+ .get(
+
operatorConfig.getFlinkClientTimeout().toSeconds(),
+ TimeUnit.SECONDS);
+
+ Map<String, String> jobConfig =
+
FlinkRuntimeConfigurationUtils.mapJobConfiguration(configurationInfo);
+ LOG.debug("Fetched {} job configuration entries",
jobConfig.size());
+ return jobConfig;
+
+ } catch (Exception e) {
+ LOG.warn("Failed to fetch job configuration via REST API", e);
Review Comment:
This catch is probably not required here as we already catch outside of this
method, this way we just hide errors unnecessarily
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java:
##########
@@ -873,6 +879,110 @@ public JobExceptionsInfoWithHistory getJobExceptions(
}
}
+ @Override
+ public Map<String, String> getJobManagerConfiguration(Configuration conf,
JobID jobId)
+ throws Exception {
+ LOG.debug("Fetching JobManager configuration");
+ try (var clusterClient = getClusterClient(conf)) {
+ var parameters = new JobMessageParameters();
+ parameters.jobPathParameter.resolve(jobId);
+
+ try {
+ var configurationInfo =
+ clusterClient
+ .sendRequest(
+
JobManagerJobConfigurationHeaders.getInstance(),
+ parameters,
+ EmptyRequestBody.getInstance())
+ .get(
+
operatorConfig.getFlinkClientTimeout().toSeconds(),
+ TimeUnit.SECONDS);
+
+ Map<String, String> jmConfig = new HashMap<>();
+ configurationInfo.forEach(entry ->
jmConfig.put(entry.getKey(), entry.getValue()));
+ LOG.debug("Fetched {} JobManager configuration entries",
jmConfig.size());
+ return jmConfig;
+
+ } catch (Exception e) {
+ LOG.warn("Failed to fetch JobManager configuration via REST
API", e);
+ return new HashMap<>();
+ }
+ }
+ }
+
+ @Override
+ public Map<String, String> getJobConfiguration(Configuration conf, JobID
jobId)
+ throws Exception {
+ LOG.debug("Fetching job configuration");
+ try (var clusterClient = getClusterClient(conf)) {
+ var jobConfigHeaders = JobConfigHeaders.getInstance();
+ var parameters = new JobMessageParameters();
+ parameters.jobPathParameter.resolve(jobId);
+
+ try {
+ JobConfigInfo configurationInfo =
+ clusterClient
+ .sendRequest(
+ jobConfigHeaders,
+ parameters,
+ EmptyRequestBody.getInstance())
+ .get(
+
operatorConfig.getFlinkClientTimeout().toSeconds(),
+ TimeUnit.SECONDS);
+
+ Map<String, String> jobConfig =
+
FlinkRuntimeConfigurationUtils.mapJobConfiguration(configurationInfo);
+ LOG.debug("Fetched {} job configuration entries",
jobConfig.size());
+ return jobConfig;
+
+ } catch (Exception e) {
+ LOG.warn("Failed to fetch job configuration via REST API", e);
+ return new HashMap<>();
+ }
+ }
+ }
+
+ @Override
+ public Map<String, String> getJobCheckpointConfiguration(Configuration
conf, JobID jobId)
+ throws Exception {
+ LOG.debug("Fetching checkpoint configuration");
+ try (var clusterClient = getClusterClient(conf)) {
+ var checkpointConfigHeaders =
CheckpointConfigHeaders.getInstance();
+ var parameters = new JobMessageParameters();
+ parameters.jobPathParameter.resolve(jobId);
+
+ try {
+ var checkpointConfigInfo =
+ clusterClient
+ .sendRequest(
+ checkpointConfigHeaders,
+ parameters,
+ EmptyRequestBody.getInstance())
+ .get(
+
operatorConfig.getFlinkClientTimeout().toSeconds(),
+ TimeUnit.SECONDS);
+
+ return
FlinkRuntimeConfigurationUtils.mapCheckpointConfiguration(
+ checkpointConfigInfo);
+
+ } catch (Exception e) {
Review Comment:
This catch is probably not required here as we already catch outside of this
method, this way we just hide errors unnecessarily
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java:
##########
@@ -873,6 +879,110 @@ public JobExceptionsInfoWithHistory getJobExceptions(
}
}
+ @Override
+ public Map<String, String> getJobManagerConfiguration(Configuration conf,
JobID jobId)
+ throws Exception {
+ LOG.debug("Fetching JobManager configuration");
+ try (var clusterClient = getClusterClient(conf)) {
+ var parameters = new JobMessageParameters();
+ parameters.jobPathParameter.resolve(jobId);
+
+ try {
+ var configurationInfo =
+ clusterClient
+ .sendRequest(
+
JobManagerJobConfigurationHeaders.getInstance(),
+ parameters,
+ EmptyRequestBody.getInstance())
+ .get(
+
operatorConfig.getFlinkClientTimeout().toSeconds(),
+ TimeUnit.SECONDS);
+
+ Map<String, String> jmConfig = new HashMap<>();
+ configurationInfo.forEach(entry ->
jmConfig.put(entry.getKey(), entry.getValue()));
+ LOG.debug("Fetched {} JobManager configuration entries",
jmConfig.size());
+ return jmConfig;
+
+ } catch (Exception e) {
+ LOG.warn("Failed to fetch JobManager configuration via REST
API", e);
+ return new HashMap<>();
+ }
+ }
+ }
+
+ @Override
+ public Map<String, String> getJobConfiguration(Configuration conf, JobID
jobId)
+ throws Exception {
+ LOG.debug("Fetching job configuration");
+ try (var clusterClient = getClusterClient(conf)) {
+ var jobConfigHeaders = JobConfigHeaders.getInstance();
+ var parameters = new JobMessageParameters();
+ parameters.jobPathParameter.resolve(jobId);
+
+ try {
+ JobConfigInfo configurationInfo =
+ clusterClient
+ .sendRequest(
+ jobConfigHeaders,
+ parameters,
+ EmptyRequestBody.getInstance())
+ .get(
+
operatorConfig.getFlinkClientTimeout().toSeconds(),
+ TimeUnit.SECONDS);
+
+ Map<String, String> jobConfig =
+
FlinkRuntimeConfigurationUtils.mapJobConfiguration(configurationInfo);
+ LOG.debug("Fetched {} job configuration entries",
jobConfig.size());
+ return jobConfig;
+
+ } catch (Exception e) {
+ LOG.warn("Failed to fetch job configuration via REST API", e);
+ return new HashMap<>();
+ }
+ }
+ }
+
+ @Override
+ public Map<String, String> getJobCheckpointConfiguration(Configuration
conf, JobID jobId)
+ throws Exception {
+ LOG.debug("Fetching checkpoint configuration");
+ try (var clusterClient = getClusterClient(conf)) {
+ var checkpointConfigHeaders =
CheckpointConfigHeaders.getInstance();
+ var parameters = new JobMessageParameters();
+ parameters.jobPathParameter.resolve(jobId);
+
+ try {
+ var checkpointConfigInfo =
+ clusterClient
+ .sendRequest(
+ checkpointConfigHeaders,
+ parameters,
+ EmptyRequestBody.getInstance())
+ .get(
+
operatorConfig.getFlinkClientTimeout().toSeconds(),
+ TimeUnit.SECONDS);
+
+ return
FlinkRuntimeConfigurationUtils.mapCheckpointConfiguration(
+ checkpointConfigInfo);
+
+ } catch (Exception e) {
+ LOG.warn("Failed to fetch checkpoint config, falling back to
job config", e);
+
+ if (e.getCause() instanceof RestClientException restException
+ && restException.getHttpResponseStatus() ==
HttpResponseStatus.NOT_FOUND) {
+ throw new RuntimeException("Job not found: " + jobId, e);
Review Comment:
I don't think we need the complex error handling here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
