You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/08/04 13:48:28 UTC
[GitHub] [flink-kubernetes-operator] SteNicholas opened a new pull request, #337: [FLINK-28763] Make savepoint format configurable for upgrades/savepoint operations
SteNicholas opened a new pull request, #337:
URL: https://github.com/apache/flink-kubernetes-operator/pull/337
## What is the purpose of the change
`SavepointFormatType` should be configurable for Flink 1.15+, therefore users can configure the savepoint format with CANONICAL/NATIVE in `spec.flinkConfiguration`.
## Brief change log
- Introduces the config option `kubernetes.operator.savepoint.format.type` to configure the `SavepointFormatType`.
## Verifying this change
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changes to the `CustomResourceDescriptors`: (yes / **no**)
- Core observer or reconciler logic that is regularly executed: (yes / **no**)
## Documentation
- Does this pull request introduce a new feature? (yes / **no**)
- If yes, how is the feature documented? (**not applicable** / docs / JavaDocs / not documented)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] gyfora commented on a diff in pull request #337: [FLINK-28763] Make savepoint format configurable for upgrades/savepoint operations
Posted by GitBox <gi...@apache.org>.
gyfora commented on code in PR #337:
URL: https://github.com/apache/flink-kubernetes-operator/pull/337#discussion_r940988337
##########
flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/NativeFlinkServiceTest.java:
##########
@@ -337,6 +339,77 @@ public void testEffectiveStatus() {
AbstractFlinkService.getEffectiveStatus(allFinished));
}
+ @Test
+ public void testNativeSavepointFormat() throws Exception {
+ final TestingClusterClient<String> testingClusterClient =
+ new TestingClusterClient<>(configuration, TestUtils.TEST_DEPLOYMENT_NAME);
+ final String savepointPath = "file:///path/of/svp";
+ final CompletableFuture<Tuple3<JobID, String, Boolean>> savepointFeature =
+ new CompletableFuture<>();
+ configuration.set(CheckpointingOptions.SAVEPOINT_DIRECTORY, savepointPath);
+ testingClusterClient.setRequestProcessor(
+ (headers, parameters, requestBody) -> {
+ savepointFeature.complete(
+ new Tuple3<>(
+ ((SavepointTriggerMessageParameters) parameters)
+ .jobID.getValue(),
+ ((SavepointTriggerRequestBody) requestBody)
+ .getTargetDirectory()
+ .get(),
+ ((SavepointTriggerRequestBody) requestBody).isCancelJob()));
+ return CompletableFuture.completedFuture(new TriggerResponse(new TriggerId()));
+ });
+ final CompletableFuture<Tuple3<JobID, Boolean, String>> stopWithSavepointFuture =
+ new CompletableFuture<>();
+ testingClusterClient.setStopWithSavepointFunction(
Review Comment:
We should improve this to also capture the format type so we can test it.
##########
flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/NativeFlinkServiceTest.java:
##########
@@ -337,6 +339,77 @@ public void testEffectiveStatus() {
AbstractFlinkService.getEffectiveStatus(allFinished));
}
+ @Test
+ public void testNativeSavepointFormat() throws Exception {
+ final TestingClusterClient<String> testingClusterClient =
+ new TestingClusterClient<>(configuration, TestUtils.TEST_DEPLOYMENT_NAME);
+ final String savepointPath = "file:///path/of/svp";
+ final CompletableFuture<Tuple3<JobID, String, Boolean>> savepointFeature =
+ new CompletableFuture<>();
+ configuration.set(CheckpointingOptions.SAVEPOINT_DIRECTORY, savepointPath);
+ testingClusterClient.setRequestProcessor(
+ (headers, parameters, requestBody) -> {
+ savepointFeature.complete(
+ new Tuple3<>(
+ ((SavepointTriggerMessageParameters) parameters)
+ .jobID.getValue(),
+ ((SavepointTriggerRequestBody) requestBody)
+ .getTargetDirectory()
+ .get(),
+ ((SavepointTriggerRequestBody) requestBody).isCancelJob()));
Review Comment:
should we also capture the savepoint format? I think thats what we want to test here
##########
flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/NativeFlinkServiceTest.java:
##########
@@ -337,6 +339,77 @@ public void testEffectiveStatus() {
AbstractFlinkService.getEffectiveStatus(allFinished));
}
+ @Test
+ public void testNativeSavepointFormat() throws Exception {
+ final TestingClusterClient<String> testingClusterClient =
+ new TestingClusterClient<>(configuration, TestUtils.TEST_DEPLOYMENT_NAME);
+ final String savepointPath = "file:///path/of/svp";
+ final CompletableFuture<Tuple3<JobID, String, Boolean>> savepointFeature =
Review Comment:
typo: `savepointFeature` -> `savepointFuture`
##########
flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/NativeFlinkServiceTest.java:
##########
@@ -337,6 +339,77 @@ public void testEffectiveStatus() {
AbstractFlinkService.getEffectiveStatus(allFinished));
}
+ @Test
+ public void testNativeSavepointFormat() throws Exception {
+ final TestingClusterClient<String> testingClusterClient =
+ new TestingClusterClient<>(configuration, TestUtils.TEST_DEPLOYMENT_NAME);
+ final String savepointPath = "file:///path/of/svp";
+ final CompletableFuture<Tuple3<JobID, String, Boolean>> savepointFeature =
+ new CompletableFuture<>();
+ configuration.set(CheckpointingOptions.SAVEPOINT_DIRECTORY, savepointPath);
+ testingClusterClient.setRequestProcessor(
+ (headers, parameters, requestBody) -> {
+ savepointFeature.complete(
+ new Tuple3<>(
+ ((SavepointTriggerMessageParameters) parameters)
+ .jobID.getValue(),
+ ((SavepointTriggerRequestBody) requestBody)
+ .getTargetDirectory()
+ .get(),
+ ((SavepointTriggerRequestBody) requestBody).isCancelJob()));
+ return CompletableFuture.completedFuture(new TriggerResponse(new TriggerId()));
+ });
+ final CompletableFuture<Tuple3<JobID, Boolean, String>> stopWithSavepointFuture =
+ new CompletableFuture<>();
+ testingClusterClient.setStopWithSavepointFunction(
+ (id, advanceToEndOfEventTime, savepointDir) -> {
+ stopWithSavepointFuture.complete(
+ new Tuple3<>(id, advanceToEndOfEventTime, savepointDir));
+ return CompletableFuture.completedFuture(savepointPath);
+ });
+
+ final FlinkService flinkService = createFlinkService(testingClusterClient);
+
+ final JobID jobID = JobID.generate();
+ final FlinkDeployment deployment = TestUtils.buildApplicationCluster();
+ deployment
+ .getSpec()
+ .getFlinkConfiguration()
+ .put(CheckpointingOptions.SAVEPOINT_DIRECTORY.key(), savepointPath);
+ deployment.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.READY);
+ JobStatus jobStatus = deployment.getStatus().getJobStatus();
+ jobStatus.setJobId(jobID.toHexString());
+ jobStatus.setState(org.apache.flink.api.common.JobStatus.RUNNING.name());
+ ReconciliationUtils.updateStatusForDeployedSpec(deployment, new Configuration());
+
+ jobStatus.setJobId(jobID.toString());
+ deployment.getStatus().setJobStatus(jobStatus);
+ flinkService.triggerSavepoint(
+ deployment.getStatus().getJobStatus().getJobId(),
+ SavepointTriggerType.MANUAL,
+ deployment.getStatus().getJobStatus().getSavepointInfo(),
+ new Configuration(configuration)
+ .set(
+ KubernetesOperatorConfigOptions.OPERATOR_SAVEPOINT_FORMAT_TYPE,
+ SavepointFormatType.NATIVE));
+ assertTrue(savepointFeature.isDone());
+ assertEquals(jobID, savepointFeature.get().f0);
+ assertEquals(savepointPath, savepointFeature.get().f1);
+ assertFalse(savepointFeature.get().f2);
+
+ flinkService.cancelJob(
+ deployment,
+ UpgradeMode.SAVEPOINT,
+ new Configuration(configManager.getObserveConfig(deployment))
+ .set(
+ KubernetesOperatorConfigOptions.OPERATOR_SAVEPOINT_FORMAT_TYPE,
+ SavepointFormatType.NATIVE));
+ assertTrue(stopWithSavepointFuture.isDone());
+ assertEquals(jobID, stopWithSavepointFuture.get().f0);
+ assertFalse(stopWithSavepointFuture.get().f1);
+ assertEquals(savepointPath, stopWithSavepointFuture.get().f2);
Review Comment:
these are nice but would be better to test the format also
##########
flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/NativeFlinkServiceTest.java:
##########
@@ -337,6 +339,77 @@ public void testEffectiveStatus() {
AbstractFlinkService.getEffectiveStatus(allFinished));
}
+ @Test
+ public void testNativeSavepointFormat() throws Exception {
+ final TestingClusterClient<String> testingClusterClient =
+ new TestingClusterClient<>(configuration, TestUtils.TEST_DEPLOYMENT_NAME);
+ final String savepointPath = "file:///path/of/svp";
+ final CompletableFuture<Tuple3<JobID, String, Boolean>> savepointFeature =
+ new CompletableFuture<>();
+ configuration.set(CheckpointingOptions.SAVEPOINT_DIRECTORY, savepointPath);
+ testingClusterClient.setRequestProcessor(
+ (headers, parameters, requestBody) -> {
+ savepointFeature.complete(
+ new Tuple3<>(
+ ((SavepointTriggerMessageParameters) parameters)
+ .jobID.getValue(),
+ ((SavepointTriggerRequestBody) requestBody)
+ .getTargetDirectory()
+ .get(),
+ ((SavepointTriggerRequestBody) requestBody).isCancelJob()));
+ return CompletableFuture.completedFuture(new TriggerResponse(new TriggerId()));
+ });
+ final CompletableFuture<Tuple3<JobID, Boolean, String>> stopWithSavepointFuture =
+ new CompletableFuture<>();
+ testingClusterClient.setStopWithSavepointFunction(
+ (id, advanceToEndOfEventTime, savepointDir) -> {
+ stopWithSavepointFuture.complete(
+ new Tuple3<>(id, advanceToEndOfEventTime, savepointDir));
+ return CompletableFuture.completedFuture(savepointPath);
+ });
+
+ final FlinkService flinkService = createFlinkService(testingClusterClient);
+
+ final JobID jobID = JobID.generate();
+ final FlinkDeployment deployment = TestUtils.buildApplicationCluster();
+ deployment
+ .getSpec()
+ .getFlinkConfiguration()
+ .put(CheckpointingOptions.SAVEPOINT_DIRECTORY.key(), savepointPath);
+ deployment.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.READY);
+ JobStatus jobStatus = deployment.getStatus().getJobStatus();
+ jobStatus.setJobId(jobID.toHexString());
+ jobStatus.setState(org.apache.flink.api.common.JobStatus.RUNNING.name());
+ ReconciliationUtils.updateStatusForDeployedSpec(deployment, new Configuration());
+
+ jobStatus.setJobId(jobID.toString());
+ deployment.getStatus().setJobStatus(jobStatus);
+ flinkService.triggerSavepoint(
+ deployment.getStatus().getJobStatus().getJobId(),
+ SavepointTriggerType.MANUAL,
+ deployment.getStatus().getJobStatus().getSavepointInfo(),
+ new Configuration(configuration)
+ .set(
+ KubernetesOperatorConfigOptions.OPERATOR_SAVEPOINT_FORMAT_TYPE,
+ SavepointFormatType.NATIVE));
+ assertTrue(savepointFeature.isDone());
+ assertEquals(jobID, savepointFeature.get().f0);
+ assertEquals(savepointPath, savepointFeature.get().f1);
+ assertFalse(savepointFeature.get().f2);
Review Comment:
these are nice but would be better to test the format also
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] SteNicholas commented on pull request #337: [FLINK-28763] Make savepoint format configurable for upgrades/savepoint operations
Posted by GitBox <gi...@apache.org>.
SteNicholas commented on PR #337:
URL: https://github.com/apache/flink-kubernetes-operator/pull/337#issuecomment-1205995421
@gyfora , sorry to miss adding the unit tests for this change. I have added the tests to verify whether to configure the `SavepointFormatType` to trigger savepoint or cancel job with different savepoint format. PTAL.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] gyfora commented on a diff in pull request #337: [FLINK-28763] Make savepoint format configurable for upgrades/savepoint operations
Posted by GitBox <gi...@apache.org>.
gyfora commented on code in PR #337:
URL: https://github.com/apache/flink-kubernetes-operator/pull/337#discussion_r938535348
##########
flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/NativeFlinkServiceTest.java:
##########
@@ -180,8 +186,10 @@ public void testCancelJobWithLastStateUpgradeMode() throws Exception {
.get());
}
- @Test
- public void testTriggerSavepoint() throws Exception {
+ @ParameterizedTest
+ @MethodSource("provideSavepointFormatType")
+ public void testTriggerSavepoint(Configuration configuration, FlinkConfigManager configManager)
+ throws Exception {
Review Comment:
These tests do not actually test that the correct savepoint format type is used during savepointing.
We should remove these parameterized tests and simply validate that if the user configures the savepoint type the rest api is called with that type in the FlinkService
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java:
##########
@@ -280,4 +281,12 @@ public class KubernetesOperatorConfigOptions {
.defaultValue(true)
.withDescription(
"Enables last-state fallback for savepoint upgrade mode. When the job is not running thus savepoint cannot be triggered but HA metadata is available for last state restore the operator can initiate the upgrade process when the flag is enabled.");
+
+ @Documentation.Section(SECTION_DYNAMIC)
+ public static final ConfigOption<String> OPERATOR_SAVEPOINT_FORMAT_TYPE =
Review Comment:
Should this be enum type instead?
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java:
##########
@@ -129,6 +131,14 @@ public static FlinkOperatorConfiguration fromConfiguration(Configuration operato
String labelSelector =
operatorConfig.getString(KubernetesOperatorConfigOptions.OPERATOR_LABEL_SELECTOR);
+ SavepointFormatType savepointFormatType =
+ SavepointFormatType.valueOf(
+ operatorConfig
+ .getString(
+ KubernetesOperatorConfigOptions
+ .OPERATOR_SAVEPOINT_FORMAT_TYPE)
+ .toUpperCase());
+
Review Comment:
This setting should not be part of the `FlinkOperatorConfiguration` is it is intended to be set on a per resource basis
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] gyfora commented on pull request #337: [FLINK-28763] Make savepoint format configurable for upgrades/savepoint operations
Posted by GitBox <gi...@apache.org>.
gyfora commented on PR #337:
URL: https://github.com/apache/flink-kubernetes-operator/pull/337#issuecomment-1205302598
Please add some tests for this before we can review it
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] gyfora merged pull request #337: [FLINK-28763] Make savepoint format configurable for upgrades/savepoint operations
Posted by GitBox <gi...@apache.org>.
gyfora merged PR #337:
URL: https://github.com/apache/flink-kubernetes-operator/pull/337
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org