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/06/28 12:47:19 UTC
[GitHub] [flink] zoltar9264 opened a new pull request, #20093: [FLINK-28172][changelog] Scatter dstl files into separate directories…
zoltar9264 opened a new pull request, #20093:
URL: https://github.com/apache/flink/pull/20093
… by job id
## What is the purpose of the change
Scatter dstl files into separate directories by job id, described in [FLINK-28172](https://issues.apache.org/jira/browse/FLINK-28172)
## Brief change log
- pass jobID to FsStateChangelogStorageFactory, FsStateChangelogStorage and StateChangeFsUploader
- make sub directory ':jobID/dstl' under base-path
## Verifying this change
This change is a trivial rework / code cleanup without any test coverage.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? 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] rkhachatryan commented on a diff in pull request #20093: [FLINK-28172][changelog] Scatter dstl files into separate directories…
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on code in PR #20093:
URL: https://github.com/apache/flink/pull/20093#discussion_r909636078
##########
flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/StateChangeFsUploader.java:
##########
@@ -83,14 +88,15 @@ public StateChangeFsUploader(
}
public StateChangeFsUploader(
+ JobID jobID,
Path basePath,
FileSystem fileSystem,
boolean compression,
int bufferSize,
ChangelogStorageMetricGroup metrics,
TaskChangelogRegistry changelogRegistry,
BiFunction<Path, Long, StreamStateHandle> handleFactory) {
- this.basePath = basePath;
+ this.basePath = new Path(basePath, String.format(DSTL_SUB_PATH, jobID.toString()));
Review Comment:
Wouldn't it be more safe to explicitly call `jobID.toHexString()` instead of `jobID.toString()`?
##########
flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/StateChangeFsUploader.java:
##########
@@ -54,6 +55,8 @@
public class StateChangeFsUploader implements StateChangeUploader {
private static final Logger LOG = LoggerFactory.getLogger(StateChangeFsUploader.class);
+ private static final String DSTL_SUB_PATH = "%s/dstl";
+
Review Comment:
Could you explain the purpose of the `/dstl` suffix here?
Isn't this `DSTL_SUB_PATH` already appended to a path where changelog files reside?
--
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] flinkbot commented on pull request #20093: [FLINK-28172][changelog] Scatter dstl files into separate directories…
Posted by GitBox <gi...@apache.org>.
flinkbot commented on PR #20093:
URL: https://github.com/apache/flink/pull/20093#issuecomment-1168684979
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "65b795f4c869927cfef6d19a9f8cc7a28dd3542e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "65b795f4c869927cfef6d19a9f8cc7a28dd3542e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 65b795f4c869927cfef6d19a9f8cc7a28dd3542e UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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] zoltar9264 commented on a diff in pull request #20093: [FLINK-28172][changelog] Scatter dstl files into separate directories…
Posted by GitBox <gi...@apache.org>.
zoltar9264 commented on code in PR #20093:
URL: https://github.com/apache/flink/pull/20093#discussion_r910588233
##########
flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/StateChangeFsUploader.java:
##########
@@ -83,14 +88,15 @@ public StateChangeFsUploader(
}
public StateChangeFsUploader(
+ JobID jobID,
Path basePath,
FileSystem fileSystem,
boolean compression,
int bufferSize,
ChangelogStorageMetricGroup metrics,
TaskChangelogRegistry changelogRegistry,
BiFunction<Path, Long, StreamStateHandle> handleFactory) {
- this.basePath = basePath;
+ this.basePath = new Path(basePath, String.format(DSTL_SUB_PATH, jobID.toString()));
Review Comment:
Thanks @rkhachatryan , I will add a unit test for verifying basePath after construction later.
--
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] zoltar9264 commented on pull request #20093: [FLINK-28172][changelog] Scatter dstl files into separate directories…
Posted by GitBox <gi...@apache.org>.
zoltar9264 commented on PR #20093:
URL: https://github.com/apache/flink/pull/20093#issuecomment-1176966940
@flinkbot run azure
--
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] zoltar9264 commented on pull request #20093: [FLINK-28172][changelog] Scatter dstl files into separate directories…
Posted by GitBox <gi...@apache.org>.
zoltar9264 commented on PR #20093:
URL: https://github.com/apache/flink/pull/20093#issuecomment-1173619223
@flinkbot run azure
--
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] zoltar9264 commented on a diff in pull request #20093: [FLINK-28172][changelog] Scatter dstl files into separate directories…
Posted by GitBox <gi...@apache.org>.
zoltar9264 commented on code in PR #20093:
URL: https://github.com/apache/flink/pull/20093#discussion_r914400421
##########
flink-dstl/flink-dstl-dfs/src/test/java/org/apache/flink/changelog/fs/StateChangeFsUploaderTest.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.changelog.fs;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.core.fs.Path;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+
+import static org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups.createUnregisteredTaskManagerJobMetricGroup;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** {@link StateChangeFsUploader} test. */
+class StateChangeFsUploaderTest {
+
+ @Test
+ void testBasePath() throws IOException {
+ JobID jobID = JobID.generate();
+ Path oriBasePath = new Path("file:///dstl-base-path");
+
+ ChangelogStorageMetricGroup metrics =
+ new ChangelogStorageMetricGroup(createUnregisteredTaskManagerJobMetricGroup());
+
+ StateChangeFsUploader uploader =
+ new StateChangeFsUploader(
+ jobID,
+ oriBasePath,
+ oriBasePath.getFileSystem(),
+ false,
+ 4096,
+ metrics,
+ TaskChangelogRegistry.NO_OP);
+
+ Path basePath = uploader.getBasePath();
+
+ assertThat(basePath.getName()).isEqualTo("dstl");
+ assertThat(basePath.getParent().getName()).isEqualTo(jobID.toHexString());
+ assertThat(basePath.getParent().getParent()).isEqualTo(oriBasePath);
Review Comment:
Thanks @rkhachatryan , you are right, this style is really better. I will take up with it. And may be we can rename PATH_SUFFIX to PATH_SUB_DIR ? I think it is more accurate, WDYT ?
--
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] rkhachatryan commented on a diff in pull request #20093: [FLINK-28172][changelog] Scatter dstl files into separate directories…
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on code in PR #20093:
URL: https://github.com/apache/flink/pull/20093#discussion_r914543657
##########
flink-dstl/flink-dstl-dfs/src/test/java/org/apache/flink/changelog/fs/StateChangeFsUploaderTest.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.changelog.fs;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.core.fs.Path;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+
+import static org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups.createUnregisteredTaskManagerJobMetricGroup;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** {@link StateChangeFsUploader} test. */
+class StateChangeFsUploaderTest {
+
+ @Test
+ void testBasePath() throws IOException {
+ JobID jobID = JobID.generate();
+ Path oriBasePath = new Path("file:///dstl-base-path");
+
+ ChangelogStorageMetricGroup metrics =
+ new ChangelogStorageMetricGroup(createUnregisteredTaskManagerJobMetricGroup());
+
+ StateChangeFsUploader uploader =
+ new StateChangeFsUploader(
+ jobID,
+ oriBasePath,
+ oriBasePath.getFileSystem(),
+ false,
+ 4096,
+ metrics,
+ TaskChangelogRegistry.NO_OP);
+
+ Path basePath = uploader.getBasePath();
+
+ assertThat(basePath.getName()).isEqualTo("dstl");
+ assertThat(basePath.getParent().getName()).isEqualTo(jobID.toHexString());
+ assertThat(basePath.getParent().getParent()).isEqualTo(oriBasePath);
Review Comment:
Thanks @zoltar9264 .
I think it makes sense.
--
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] rkhachatryan commented on a diff in pull request #20093: [FLINK-28172][changelog] Scatter dstl files into separate directories…
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on code in PR #20093:
URL: https://github.com/apache/flink/pull/20093#discussion_r914204861
##########
flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/StateChangeFsUploader.java:
##########
@@ -54,6 +55,8 @@
public class StateChangeFsUploader implements StateChangeUploader {
private static final Logger LOG = LoggerFactory.getLogger(StateChangeFsUploader.class);
+ private static final String DSTL_SUB_PATH = "%s/dstl";
+
Review Comment:
Thanks for the clarification.
--
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] zoltar9264 commented on pull request #20093: [FLINK-28172][changelog] Scatter dstl files into separate directories…
Posted by GitBox <gi...@apache.org>.
zoltar9264 commented on PR #20093:
URL: https://github.com/apache/flink/pull/20093#issuecomment-1174525564
@flinkbot run azure
--
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] zoltar9264 commented on a diff in pull request #20093: [FLINK-28172][changelog] Scatter dstl files into separate directories…
Posted by GitBox <gi...@apache.org>.
zoltar9264 commented on code in PR #20093:
URL: https://github.com/apache/flink/pull/20093#discussion_r914398852
##########
flink-dstl/flink-dstl-dfs/src/test/java/org/apache/flink/changelog/fs/StateChangeFsUploaderTest.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.changelog.fs;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.core.fs.Path;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+
+import static org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups.createUnregisteredTaskManagerJobMetricGroup;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** {@link StateChangeFsUploader} test. */
+class StateChangeFsUploaderTest {
+
+ @Test
+ void testBasePath() throws IOException {
+ JobID jobID = JobID.generate();
+ Path oriBasePath = new Path("file:///dstl-base-path");
+
+ ChangelogStorageMetricGroup metrics =
+ new ChangelogStorageMetricGroup(createUnregisteredTaskManagerJobMetricGroup());
+
+ StateChangeFsUploader uploader =
+ new StateChangeFsUploader(
+ jobID,
+ oriBasePath,
+ oriBasePath.getFileSystem(),
+ false,
+ 4096,
+ metrics,
+ TaskChangelogRegistry.NO_OP);
+
+ Path basePath = uploader.getBasePath();
+
+ assertThat(basePath.getName()).isEqualTo("dstl");
Review Comment:
Thanks @rkhachatryan , I will fix 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] zoltar9264 commented on a diff in pull request #20093: [FLINK-28172][changelog] Scatter dstl files into separate directories…
Posted by GitBox <gi...@apache.org>.
zoltar9264 commented on code in PR #20093:
URL: https://github.com/apache/flink/pull/20093#discussion_r910588233
##########
flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/StateChangeFsUploader.java:
##########
@@ -83,14 +88,15 @@ public StateChangeFsUploader(
}
public StateChangeFsUploader(
+ JobID jobID,
Path basePath,
FileSystem fileSystem,
boolean compression,
int bufferSize,
ChangelogStorageMetricGroup metrics,
TaskChangelogRegistry changelogRegistry,
BiFunction<Path, Long, StreamStateHandle> handleFactory) {
- this.basePath = basePath;
+ this.basePath = new Path(basePath, String.format(DSTL_SUB_PATH, jobID.toString()));
Review Comment:
Thanks @rkhachatryan , I will add a unit test for verify basePath after construction later.
--
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] zoltar9264 commented on a diff in pull request #20093: [FLINK-28172][changelog] Scatter dstl files into separate directories…
Posted by GitBox <gi...@apache.org>.
zoltar9264 commented on code in PR #20093:
URL: https://github.com/apache/flink/pull/20093#discussion_r910587799
##########
flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/StateChangeFsUploader.java:
##########
@@ -83,14 +88,15 @@ public StateChangeFsUploader(
}
public StateChangeFsUploader(
+ JobID jobID,
Path basePath,
FileSystem fileSystem,
boolean compression,
int bufferSize,
ChangelogStorageMetricGroup metrics,
TaskChangelogRegistry changelogRegistry,
BiFunction<Path, Long, StreamStateHandle> handleFactory) {
- this.basePath = basePath;
+ this.basePath = new Path(basePath, String.format(DSTL_SUB_PATH, jobID.toString()));
Review Comment:
Yes, you are right, I will fix it later.
--
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] zoltar9264 commented on pull request #20093: [FLINK-28172][changelog] Scatter dstl files into separate directories…
Posted by GitBox <gi...@apache.org>.
zoltar9264 commented on PR #20093:
URL: https://github.com/apache/flink/pull/20093#issuecomment-1175010353
@flinkbot run azure
--
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] rkhachatryan commented on a diff in pull request #20093: [FLINK-28172][changelog] Scatter dstl files into separate directories…
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on code in PR #20093:
URL: https://github.com/apache/flink/pull/20093#discussion_r914214042
##########
flink-dstl/flink-dstl-dfs/src/test/java/org/apache/flink/changelog/fs/StateChangeFsUploaderTest.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.changelog.fs;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.core.fs.Path;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+
+import static org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups.createUnregisteredTaskManagerJobMetricGroup;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** {@link StateChangeFsUploader} test. */
+class StateChangeFsUploaderTest {
+
+ @Test
+ void testBasePath() throws IOException {
+ JobID jobID = JobID.generate();
+ Path oriBasePath = new Path("file:///dstl-base-path");
+
+ ChangelogStorageMetricGroup metrics =
+ new ChangelogStorageMetricGroup(createUnregisteredTaskManagerJobMetricGroup());
+
+ StateChangeFsUploader uploader =
+ new StateChangeFsUploader(
+ jobID,
+ oriBasePath,
+ oriBasePath.getFileSystem(),
+ false,
+ 4096,
+ metrics,
+ TaskChangelogRegistry.NO_OP);
+
+ Path basePath = uploader.getBasePath();
+
+ assertThat(basePath.getName()).isEqualTo("dstl");
+ assertThat(basePath.getParent().getName()).isEqualTo(jobID.toHexString());
+ assertThat(basePath.getParent().getParent()).isEqualTo(oriBasePath);
Review Comment:
NIT: I'm wondering if something like
```
assertEquals(
uploader.getBasePath().getPath(),
String.format(
"/%s/%s/%s",
rootPath, jobID.toHexString(), StateChangeFsUploader.PATH_SUFFIX));
```
will be more readable, WDYT?
##########
flink-dstl/flink-dstl-dfs/src/test/java/org/apache/flink/changelog/fs/StateChangeFsUploaderTest.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.changelog.fs;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.core.fs.Path;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+
+import static org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups.createUnregisteredTaskManagerJobMetricGroup;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** {@link StateChangeFsUploader} test. */
+class StateChangeFsUploaderTest {
+
+ @Test
+ void testBasePath() throws IOException {
+ JobID jobID = JobID.generate();
+ Path oriBasePath = new Path("file:///dstl-base-path");
+
+ ChangelogStorageMetricGroup metrics =
+ new ChangelogStorageMetricGroup(createUnregisteredTaskManagerJobMetricGroup());
+
+ StateChangeFsUploader uploader =
+ new StateChangeFsUploader(
+ jobID,
+ oriBasePath,
+ oriBasePath.getFileSystem(),
+ false,
+ 4096,
+ metrics,
+ TaskChangelogRegistry.NO_OP);
+
+ Path basePath = uploader.getBasePath();
+
+ assertThat(basePath.getName()).isEqualTo("dstl");
Review Comment:
I'd extract `"dstl"` constant in `StateChangeFsUploader` and use it here, so that changes are reflected in test.
--
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] zoltar9264 commented on pull request #20093: [FLINK-28172][changelog] Scatter dstl files into separate directories…
Posted by GitBox <gi...@apache.org>.
zoltar9264 commented on PR #20093:
URL: https://github.com/apache/flink/pull/20093#issuecomment-1169447489
@flinkbot run azure
--
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] rkhachatryan merged pull request #20093: [FLINK-28172][changelog] Scatter dstl files into separate directories…
Posted by GitBox <gi...@apache.org>.
rkhachatryan merged PR #20093:
URL: https://github.com/apache/flink/pull/20093
--
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] zoltar9264 commented on a diff in pull request #20093: [FLINK-28172][changelog] Scatter dstl files into separate directories…
Posted by GitBox <gi...@apache.org>.
zoltar9264 commented on code in PR #20093:
URL: https://github.com/apache/flink/pull/20093#discussion_r910585869
##########
flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/StateChangeFsUploader.java:
##########
@@ -54,6 +55,8 @@
public class StateChangeFsUploader implements StateChangeUploader {
private static final Logger LOG = LoggerFactory.getLogger(StateChangeFsUploader.class);
+ private static final String DSTL_SUB_PATH = "%s/dstl";
+
Review Comment:
This purpose is described in the last paragraph of the [FLINK-28172](https://issues.apache.org/jira/browse/FLINK-28172).
> Going a step further, we can even create two levels of subdirectories under the dstl.dfs.base-path directory, like base-path/{jobId}/dstl . This way, if the user configures the same dstl.dfs.base-path as state.checkpoints.dir, all files needed for job recovery will be in the same directory and well organized.
WDYT ?
--
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] zoltar9264 commented on pull request #20093: [FLINK-28172][changelog] Scatter dstl files into separate directories…
Posted by GitBox <gi...@apache.org>.
zoltar9264 commented on PR #20093:
URL: https://github.com/apache/flink/pull/20093#issuecomment-1178453865
Thanks @rkhachatryan . Done.
--
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