[FLINK-35265][snapshot] Create FlinkStateSnapshot in the namespace of job

Author: mateczagany

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkStateSnapshotUtils.java

@@ -122,10 +122,12 @@ public static String getValidatedFlinkStateSnapshotPath(
 
     protected static FlinkStateSnapshot createFlinkStateSnapshot(
             KubernetesClient kubernetesClient,
+            String namespace,
             String name,
             FlinkStateSnapshotSpec spec,
             SnapshotTriggerType triggerType) {
         var metadata = new ObjectMeta();
+        metadata.setNamespace(namespace);
         metadata.setName(name);
         metadata.getLabels().put(CrdConstants.LABEL_SNAPSHOT_TYPE, triggerType.name());
 
@@ -169,7 +171,12 @@ public static FlinkStateSnapshot createSavepointResource(
                         .build();
 
         var resourceName = getFlinkStateSnapshotName(SAVEPOINT, triggerType, resource);
-        return createFlinkStateSnapshot(kubernetesClient, resourceName, snapshotSpec, triggerType);
+        return createFlinkStateSnapshot(
+                kubernetesClient,
+                resource.getMetadata().getNamespace(),
+                resourceName,
+                snapshotSpec,
+                triggerType);
     }
 
     /**
@@ -191,7 +198,12 @@ public static FlinkStateSnapshot createCheckpointResource(
                         .build();
 
         var resourceName = getFlinkStateSnapshotName(CHECKPOINT, triggerType, resource);
-        return createFlinkStateSnapshot(kubernetesClient, resourceName, snapshotSpec, triggerType);
+        return createFlinkStateSnapshot(
+                kubernetesClient,
+                resource.getMetadata().getNamespace(),
+                resourceName,
+                snapshotSpec,
+                triggerType);
     }
 
     /**

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java

@@ -286,7 +286,12 @@ public KubernetesClient getClient() {
     public static <CR extends AbstractFlinkResource<?, ?>>
             List<FlinkStateSnapshot> getFlinkStateSnapshotsForResource(
                     KubernetesClient kubernetesClient, CR resource) {
-        return kubernetesClient.resources(FlinkStateSnapshot.class).list().getItems().stream()
+        return kubernetesClient
+                .resources(FlinkStateSnapshot.class)
+                .inAnyNamespace()
+                .list()
+                .getItems()
+                .stream()
                 .filter(
                         s ->
                                 s.getSpec()

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/FlinkStateSnapshotUtilsTest.java

@@ -229,6 +229,28 @@ public void testCreatePeriodicSavepointResource() {
                 false);
     }
 
+    @Test
+    public void testCreateSnapshotInSameNamespace() {
+        var namespace = "different-namespace";
+        var deployment = initDeployment();
+        deployment.getMetadata().setNamespace(namespace);
+
+        var savepoint =
+                FlinkStateSnapshotUtils.createSavepointResource(
+                        client,
+                        deployment,
+                        SAVEPOINT_PATH,
+                        SnapshotTriggerType.PERIODIC,
+                        SavepointFormatType.CANONICAL,
+                        true);
+        assertThat(savepoint.getMetadata().getNamespace()).isEqualTo(namespace);
+
+        var checkpoint =
+                FlinkStateSnapshotUtils.createCheckpointResource(
+                        client, deployment, SnapshotTriggerType.MANUAL);
+        assertThat(checkpoint.getMetadata().getNamespace()).isEqualTo(namespace);
+    }
+
     @Test
     public void testCreateCheckpointResource() {
         var deployment = initDeployment();