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/01/27 15:49:10 UTC
[GitHub] [flink] autophagy opened a new pull request #18543: [FLINK-25431] Implement a file-based JobResultStore
autophagy opened a new pull request #18543:
URL: https://github.com/apache/flink/pull/18543
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
This pull request contains an implementation of the `JobResultStore` backed by a file-system, and attempts to fulfil the definition provided by [FLIP-194](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=195726435#FLIP194:IntroducetheJobResultStore-FileSystemJobResultStore). By default, this filesystem is based on the settings for `high-availability.storageDir` and the cluster-id, but can be overriden with the `job-result-store.storage-path setting`.
Alongside this setting, it also introduces a new option `job-result-store.delete-on-commit`, which determines whether the underlying file for the job result entry is deleted when it is clean, or whether it is renamed and persisted. By default, this option is true.
This PR also sets the various high-availability services, such as Kubernetes and Zookeeper, to use this implementation by default. `EmbeddedJobResultStore` is now only used by `AbstractNonHaServices`.
## Brief change log
- *Added an file-system based Job Result Store to be used, by default, in high-availability settings.*
## Verifying this change
Please make sure both new and modified tests in this PR follows the conventions defined in our code quality guide: https://flink.apache.org/contributing/code-style-and-quality-common.html#testing
This change added tests and can be verified as follows:
*(example:)*
- *Added unit tests that ensure the Filesystem Job Result Store fulfils the contract specified by the interface.*
- *Added unit tests that ensure the Filesystem Job Result Store creates/deletes/renames files as expected according to the FLIP.
## 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? (yes)
- If yes, how is the feature documented? (docs / JavaDocs)
--
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] XComp commented on a change in pull request #18543: [FLINK-25431] Implement a file-based JobResultStore
Posted by GitBox <gi...@apache.org>.
XComp commented on a change in pull request #18543:
URL: https://github.com/apache/flink/pull/18543#discussion_r794271632
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/JobResultStoreContractTest.java
##########
@@ -0,0 +1,163 @@
+/*
+ * 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.runtime.highavailability;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.jobmaster.JobResult;
+import org.apache.flink.runtime.testutils.TestingJobResultStore;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.NoSuchElementException;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
+
+/**
+ * This interface defines a series of tests for any implementation of the {@link JobResultStore} to
+ * determine whether they correctly implement the contracts defined by the interface.
+ */
+public interface JobResultStoreContractTest {
+
+ JobResultEntry DUMMY_JOB_RESULT_ENTRY =
+ new JobResultEntry(TestingJobResultStore.DUMMY_JOB_RESULT);
+
+ JobResultStore createJobResultStore() throws IOException;
+
+ @Test
+ default void testStoreDirtyJobResult() throws IOException {
+ JobResultStore jobResultStore = createJobResultStore();
+ jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY);
+ assertThat(jobResultStore.hasDirtyJobResultEntry(DUMMY_JOB_RESULT_ENTRY.getJobId()))
+ .isTrue();
+ }
Review comment:
just to add more context: Here, the motivation is to have less code that needs to be maintained. In an ideal world we would like to have each code path tested through a single test. That has the benefit of only changing a single test if that specific codepath was adapted.
--
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 #18543: [FLINK-25431] Implement a file-based JobResultStore
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #18543:
URL: https://github.com/apache/flink/pull/18543#issuecomment-1023364026
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2cfc588752f0396b9c2d862585da6d7c2c46fb38",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "2cfc588752f0396b9c2d862585da6d7c2c46fb38",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2cfc588752f0396b9c2d862585da6d7c2c46fb38 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] flinkbot commented on pull request #18543: [FLINK-25431] Implement a file-based JobResultStore
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #18543:
URL: https://github.com/apache/flink/pull/18543#issuecomment-1023362541
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit 2cfc588752f0396b9c2d862585da6d7c2c46fb38 (Thu Jan 27 15:54:52 UTC 2022)
**Warnings:**
* Documentation files were touched, but no `docs/content.zh/` files: Update Chinese documentation or file Jira ticket.
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</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] flinkbot edited a comment on pull request #18543: [FLINK-25431] Implement a file-based JobResultStore
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18543:
URL: https://github.com/apache/flink/pull/18543#issuecomment-1023364026
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2cfc588752f0396b9c2d862585da6d7c2c46fb38",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30329",
"triggerID" : "2cfc588752f0396b9c2d862585da6d7c2c46fb38",
"triggerType" : "PUSH"
}, {
"hash" : "0d2150b89e191ae31e3bca29aed90d0508bc002a",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30380",
"triggerID" : "0d2150b89e191ae31e3bca29aed90d0508bc002a",
"triggerType" : "PUSH"
}, {
"hash" : "6e5a51ffbc7bbe8145a6f78d2e0c4f49bdcdc6ea",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "6e5a51ffbc7bbe8145a6f78d2e0c4f49bdcdc6ea",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2cfc588752f0396b9c2d862585da6d7c2c46fb38 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30329)
* 0d2150b89e191ae31e3bca29aed90d0508bc002a Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30380)
* 6e5a51ffbc7bbe8145a6f78d2e0c4f49bdcdc6ea 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] XComp commented on a change in pull request #18543: [FLINK-25431] Implement a file-based JobResultStore
Posted by GitBox <gi...@apache.org>.
XComp commented on a change in pull request #18543:
URL: https://github.com/apache/flink/pull/18543#discussion_r793813631
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/JobResultStoreOptions.java
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.runtime.highavailability;
+
+import org.apache.flink.annotation.docs.Documentation;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.HighAvailabilityOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.configuration.description.TextElement;
+
+/** The set of configuration options relating to the Job Result Store. */
+public class JobResultStoreOptions {
+ @Documentation.Section(Documentation.Sections.COMMON_HIGH_AVAILABILITY_JOB_RESULT_STORE)
+ public static final ConfigOption<String> STORAGE_PATH =
+ ConfigOptions.key("job-result-store.storage-path")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Defines where job results should be stored. This should be an "
+ + "underlying file-system that provides read-after-write consistency. By "
+ + "default, this is %s.",
+ TextElement.code(
+ FileSystemJobResultStore
+ .createDefaultJobResultStorePath(
+ HighAvailabilityOptions
+ .HA_STORAGE_PATH
+ .key(),
+ HighAvailabilityOptions
+ .HA_CLUSTER_ID
+ .key())))
+ .build());
+
+ @Documentation.Section(Documentation.Sections.COMMON_HIGH_AVAILABILITY_JOB_RESULT_STORE)
+ public static final ConfigOption<Boolean> DELETE_ON_COMMIT =
+ ConfigOptions.key("job-result-store.delete-on-commit")
+ .booleanType()
+ .defaultValue(Boolean.TRUE)
+ .withDescription(
+ "Determines whether job results should be automatically removed "
+ + "from the underlying job result store when they are clean. If "
Review comment:
```suggestion
+ "from the underlying job result store when the corresponding entry was transitioned into clean state. If "
```
nit: that one I just added after noticing the other typo below. ...otherwise, I would have left it like that. 😇
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/JobResultStoreOptions.java
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.runtime.highavailability;
+
+import org.apache.flink.annotation.docs.Documentation;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.HighAvailabilityOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.configuration.description.TextElement;
+
+/** The set of configuration options relating to the Job Result Store. */
+public class JobResultStoreOptions {
+ @Documentation.Section(Documentation.Sections.COMMON_HIGH_AVAILABILITY_JOB_RESULT_STORE)
+ public static final ConfigOption<String> STORAGE_PATH =
+ ConfigOptions.key("job-result-store.storage-path")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Defines where job results should be stored. This should be an "
+ + "underlying file-system that provides read-after-write consistency. By "
+ + "default, this is %s.",
+ TextElement.code(
+ FileSystemJobResultStore
+ .createDefaultJobResultStorePath(
+ HighAvailabilityOptions
+ .HA_STORAGE_PATH
+ .key(),
+ HighAvailabilityOptions
+ .HA_CLUSTER_ID
+ .key())))
+ .build());
+
+ @Documentation.Section(Documentation.Sections.COMMON_HIGH_AVAILABILITY_JOB_RESULT_STORE)
+ public static final ConfigOption<Boolean> DELETE_ON_COMMIT =
+ ConfigOptions.key("job-result-store.delete-on-commit")
+ .booleanType()
+ .defaultValue(Boolean.TRUE)
+ .withDescription(
+ "Determines whether job results should be automatically removed "
+ + "from the underlying job result store when they are clean. If "
+ + "false, the cleaned job results are, instead, marked as clean "
+ + "to indicate their state. In this case, Flink no longer has"
+ + "ownership and the resources need to be cleaned up by the"
Review comment:
```suggestion
+ "ownership and the resources need to be cleaned up by the "
```
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/JobResultStoreOptions.java
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.runtime.highavailability;
+
+import org.apache.flink.annotation.docs.Documentation;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.HighAvailabilityOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.configuration.description.TextElement;
+
+/** The set of configuration options relating to the Job Result Store. */
+public class JobResultStoreOptions {
+ @Documentation.Section(Documentation.Sections.COMMON_HIGH_AVAILABILITY_JOB_RESULT_STORE)
+ public static final ConfigOption<String> STORAGE_PATH =
+ ConfigOptions.key("job-result-store.storage-path")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Defines where job results should be stored. This should be an "
+ + "underlying file-system that provides read-after-write consistency. By "
+ + "default, this is %s.",
+ TextElement.code(
+ FileSystemJobResultStore
+ .createDefaultJobResultStorePath(
+ HighAvailabilityOptions
+ .HA_STORAGE_PATH
+ .key(),
+ HighAvailabilityOptions
+ .HA_CLUSTER_ID
+ .key())))
+ .build());
+
+ @Documentation.Section(Documentation.Sections.COMMON_HIGH_AVAILABILITY_JOB_RESULT_STORE)
+ public static final ConfigOption<Boolean> DELETE_ON_COMMIT =
+ ConfigOptions.key("job-result-store.delete-on-commit")
+ .booleanType()
+ .defaultValue(Boolean.TRUE)
+ .withDescription(
+ "Determines whether job results should be automatically removed "
+ + "from the underlying job result store when they are clean. If "
+ + "false, the cleaned job results are, instead, marked as clean "
+ + "to indicate their state. In this case, Flink no longer has"
Review comment:
```suggestion
+ "to indicate their state. In this case, Flink no longer has "
```
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedJobResultStoreTestInternal.java
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.runtime.highavailability.nonha.embedded;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.highavailability.JobResultEntry;
+import org.apache.flink.runtime.testutils.TestingJobResultStore;
+import org.apache.flink.util.TestLoggerExtension;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.AssertionsForClassTypes.entry;
+
+/** Tests for the internal {@link EmbeddedJobResultStore} mechanisms. */
+@ExtendWith(TestLoggerExtension.class)
+public class EmbeddedJobResultStoreTestInternal {
+
+ private static final JobResultEntry DUMMY_JOB_RESULT_ENTRY =
Review comment:
What about making the `JobResultStoreContractTest.DUMM_JOB_RESULT_ENTRY` public an using it here as well?
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedJobResultStoreTestInternal.java
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.runtime.highavailability.nonha.embedded;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.highavailability.JobResultEntry;
+import org.apache.flink.runtime.testutils.TestingJobResultStore;
+import org.apache.flink.util.TestLoggerExtension;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.AssertionsForClassTypes.entry;
+
+/** Tests for the internal {@link EmbeddedJobResultStore} mechanisms. */
+@ExtendWith(TestLoggerExtension.class)
+public class EmbeddedJobResultStoreTestInternal {
Review comment:
It feels like this test class is not really necessary anymore. We're covering all the stuff in `JobResultStoreContractTest` already. Testing the internal representation is not necessary because it's not user-facing (in contrast to the `FileSystemJobResultStore` where the user is interacting with the folder in some usecases.
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/JobResultStoreContractTest.java
##########
@@ -0,0 +1,163 @@
+/*
+ * 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.runtime.highavailability;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.jobmaster.JobResult;
+import org.apache.flink.runtime.testutils.TestingJobResultStore;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.NoSuchElementException;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
+
+/**
+ * This interface defines a series of tests for any implementation of the {@link JobResultStore} to
+ * determine whether they correctly implement the contracts defined by the interface.
+ */
+public interface JobResultStoreContractTest {
+
+ JobResultEntry DUMMY_JOB_RESULT_ENTRY =
+ new JobResultEntry(TestingJobResultStore.DUMMY_JOB_RESULT);
+
+ JobResultStore createJobResultStore() throws IOException;
+
+ @Test
+ default void testStoreDirtyJobResult() throws IOException {
+ JobResultStore jobResultStore = createJobResultStore();
+ jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY);
+ assertThat(jobResultStore.hasDirtyJobResultEntry(DUMMY_JOB_RESULT_ENTRY.getJobId()))
+ .isTrue();
+ }
+
+ @Test
+ default void testStoreDuplicateJobResultThrowsException() throws IOException {
+ JobResultStore jobResultStore = createJobResultStore();
+ jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY);
+ assertThatThrownBy(() -> jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY))
+ .isInstanceOf(IllegalStateException.class);
+ }
+
+ @Test
+ default void testStoreCleanedJobResultThrowsException() throws IOException {
+ JobResultStore jobResultStore = createJobResultStore();
+ jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY);
+ jobResultStore.markResultAsClean(DUMMY_JOB_RESULT_ENTRY.getJobId());
+ assertThatThrownBy(() -> jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY))
+ .isInstanceOf(IllegalStateException.class);
+ }
+
+ @Test
+ default void testMarkDirtyResultAsClean() throws IOException {
Review comment:
Is obsolete: It's already covered by `testHasJobResultEntryWithCleanEntry`
##########
File path: docs/layouts/shortcodes/generated/common_high_availability_jrs_section.html
##########
@@ -0,0 +1,24 @@
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Key</th>
+ <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>job-result-store.delete-on-commit</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Determines whether job results should be automatically removed from the underlying job result store when they are clean. If false, the cleaned job results are, instead, marked as clean to indicate their state. In this case, Flink no longer hasownership and the resources need to be cleaned up by theuser.</td>
+ </tr>
+ <tr>
+ <td><h5>job-result-store.storage-path</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Defines where job results should be stored. This should be an underlying file-system that provides read-after-write consistency. By default, this is <code class="highlighter-rouge">high-availability.storageDir/job-result-store/high-availability.cluster-id</code>.</td>
Review comment:
```suggestion
<td>Defines where job results should be stored. This should be an underlying file-system that provides read-after-write consistency. By default, this is <code class="highlighter-rouge">high-availability.storageDir/job-result-store/high-availability.cluster-id</code>.</td>
```
The issue is now that it's difficult to differentiate plain text from variables in the default path. That's why I proposed `FileSystemJobResultStore.createDefaultJobResultStorePath(String.format("{%s}", HighAvailabilityOptions.HA_STORAGE_PATH.key()), String.format("{%s}", HighAvailabilityOptions.HA_CLUSTER_ID.key()))))`.
I'm wondering whether we should add a generic `ConfigOption.keyInPathRepresentation` method to `ConfigOption` to make this available as a consistent format within also for other descriptions. But it feels like it's not the right place to do this because `ConfigOption` is actually not meant for formatting information. 🤔
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/AbstractThreadsafeJobResultStore.java
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.runtime.highavailability;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.jobmaster.JobResult;
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.IOException;
+import java.util.NoSuchElementException;
+import java.util.Set;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/** An abstract class for threadsafe implementations of the {@link JobResultStore}. */
+public abstract class AbstractThreadsafeJobResultStore implements JobResultStore {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(AbstractThreadsafeJobResultStore.class);
+
+ private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
+
+ @Override
+ public void createDirtyResult(JobResultEntry jobResultEntry) throws IOException {
+ Preconditions.checkState(
+ !hasJobResultEntry(jobResultEntry.getJobId()),
+ "Job result store already contains an entry for for job %s",
+ jobResultEntry.getJobId());
+
+ readWriteLock.writeLock().lock();
+ try {
+ createDirtyResultInternal(jobResultEntry);
+ } finally {
+ readWriteLock.writeLock().unlock();
+ }
+ }
+
+ @GuardedBy("readWriteLock")
+ protected abstract void createDirtyResultInternal(JobResultEntry jobResultEntry)
+ throws IOException;
+
+ @Override
+ public void markResultAsClean(JobID jobId) throws IOException, NoSuchElementException {
+ if (hasCleanJobResultEntry(jobId)) {
+ LOG.debug("The job {} is already marked as clean. No action required.", jobId);
+ return;
+ }
+
+ readWriteLock.writeLock().lock();
+ try {
+ markResultAsCleanInternal(jobId);
+ } finally {
+ readWriteLock.writeLock().unlock();
+ }
+ }
+
+ @GuardedBy("readWriteLock")
+ protected abstract void markResultAsCleanInternal(JobID jobId)
+ throws IOException, NoSuchElementException;
+
+ @Override
+ public boolean hasJobResultEntry(JobID jobId) throws IOException {
+ readWriteLock.readLock().lock();
+ try {
+ return hasDirtyJobResultEntryInternal(jobId) || hasCleanJobResultEntryInternal(jobId);
+ } finally {
+ readWriteLock.readLock().unlock();
+ }
+ }
+
+ @Override
+ public boolean hasDirtyJobResultEntry(JobID jobId) throws IOException {
+ readWriteLock.readLock().lock();
+ try {
+ return hasDirtyJobResultEntryInternal(jobId);
+ } finally {
+ readWriteLock.readLock().unlock();
+ }
+ }
+
+ @GuardedBy("readWriteLock")
+ protected abstract boolean hasDirtyJobResultEntryInternal(JobID jobId) throws IOException;
+
+ @Override
+ public boolean hasCleanJobResultEntry(JobID jobId) throws IOException {
+ readWriteLock.readLock().lock();
+ try {
+ return hasCleanJobResultEntryInternal(jobId);
+ } finally {
+ readWriteLock.readLock().unlock();
+ }
+ }
+
+ @GuardedBy("readWriteLock")
+ protected abstract boolean hasCleanJobResultEntryInternal(JobID jobId) throws IOException;
+
+ @Override
+ public Set<JobResult> getDirtyResults() throws IOException {
+ readWriteLock.readLock().lock();
+ try {
+ return getDirtyResultsInternal();
+ } finally {
+ readWriteLock.readLock().unlock();
+ }
+ }
+
+ @GuardedBy("readWriteLock")
+ protected abstract Set<JobResult> getDirtyResultsInternal() throws IOException;
Review comment:
```suggestion
protected abstract Set<JobResult> getDirtyResultsInternal() throws IOException;
private void withWriteLock(ThrowingRunnable<IOException> runnable) throws IOException {
readWriteLock.writeLock().lock();
try {
runnable.run();
} finally {
readWriteLock.writeLock().unlock();
}
}
private <T> T withReadLock(SupplierWithException<T, IOException> runnable) throws IOException {
readWriteLock.readLock().lock();
try {
return runnable.get();
} finally {
readWriteLock.readLock().unlock();
}
}
```
Sorry for not doing it earlier but I looked once more into what you meant by the lock helper methods not being suitable due to the `IOException`. You're right that we cannot use the implementation from the `EmbeddedJobResultStore` right-away. But we could us `SupplierWithException` and `ThrowingRunnable<IOException>`. This would forward the IOException through the method signature and enables us to have dedicated methods for utilizing the read and the write lock.
That's another hint: All the default callback interfaces (e.g. `Runnable`, `Function`, `Supplier`, `Consumer`, ...) have corresponding implementations considering exceptions (e.g. `ThrowingRunnable`, `SupplierWithException`).
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/JobResultStoreContractTest.java
##########
@@ -0,0 +1,163 @@
+/*
+ * 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.runtime.highavailability;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.jobmaster.JobResult;
+import org.apache.flink.runtime.testutils.TestingJobResultStore;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.NoSuchElementException;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
+
+/**
+ * This interface defines a series of tests for any implementation of the {@link JobResultStore} to
+ * determine whether they correctly implement the contracts defined by the interface.
+ */
+public interface JobResultStoreContractTest {
+
+ JobResultEntry DUMMY_JOB_RESULT_ENTRY =
+ new JobResultEntry(TestingJobResultStore.DUMMY_JOB_RESULT);
+
+ JobResultStore createJobResultStore() throws IOException;
+
+ @Test
+ default void testStoreDirtyJobResult() throws IOException {
+ JobResultStore jobResultStore = createJobResultStore();
+ jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY);
+ assertThat(jobResultStore.hasDirtyJobResultEntry(DUMMY_JOB_RESULT_ENTRY.getJobId()))
+ .isTrue();
+ }
+
+ @Test
+ default void testStoreDuplicateJobResultThrowsException() throws IOException {
+ JobResultStore jobResultStore = createJobResultStore();
+ jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY);
+ assertThatThrownBy(() -> jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY))
+ .isInstanceOf(IllegalStateException.class);
+ }
+
+ @Test
+ default void testStoreCleanedJobResultThrowsException() throws IOException {
Review comment:
```suggestion
default void testAddDirtyEntryForAlreadyCleanedJobResultThrowsException() throws IOException {
```
The other name was misleading
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/JobResultStoreContractTest.java
##########
@@ -0,0 +1,163 @@
+/*
+ * 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.runtime.highavailability;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.jobmaster.JobResult;
+import org.apache.flink.runtime.testutils.TestingJobResultStore;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.NoSuchElementException;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
+
+/**
+ * This interface defines a series of tests for any implementation of the {@link JobResultStore} to
+ * determine whether they correctly implement the contracts defined by the interface.
+ */
+public interface JobResultStoreContractTest {
+
+ JobResultEntry DUMMY_JOB_RESULT_ENTRY =
+ new JobResultEntry(TestingJobResultStore.DUMMY_JOB_RESULT);
+
+ JobResultStore createJobResultStore() throws IOException;
+
+ @Test
+ default void testStoreDirtyJobResult() throws IOException {
+ JobResultStore jobResultStore = createJobResultStore();
+ jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY);
+ assertThat(jobResultStore.hasDirtyJobResultEntry(DUMMY_JOB_RESULT_ENTRY.getJobId()))
+ .isTrue();
+ }
Review comment:
That one can be removed. It's covered by `testHasJobResultEntryWithDirtyEntry`
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/JobResultStoreContractTest.java
##########
@@ -0,0 +1,163 @@
+/*
+ * 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.runtime.highavailability;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.jobmaster.JobResult;
+import org.apache.flink.runtime.testutils.TestingJobResultStore;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.NoSuchElementException;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
+
+/**
+ * This interface defines a series of tests for any implementation of the {@link JobResultStore} to
+ * determine whether they correctly implement the contracts defined by the interface.
+ */
+public interface JobResultStoreContractTest {
+
+ JobResultEntry DUMMY_JOB_RESULT_ENTRY =
+ new JobResultEntry(TestingJobResultStore.DUMMY_JOB_RESULT);
+
+ JobResultStore createJobResultStore() throws IOException;
+
+ @Test
+ default void testStoreDirtyJobResult() throws IOException {
+ JobResultStore jobResultStore = createJobResultStore();
+ jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY);
+ assertThat(jobResultStore.hasDirtyJobResultEntry(DUMMY_JOB_RESULT_ENTRY.getJobId()))
+ .isTrue();
+ }
+
+ @Test
+ default void testStoreDuplicateJobResultThrowsException() throws IOException {
+ JobResultStore jobResultStore = createJobResultStore();
+ jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY);
+ assertThatThrownBy(() -> jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY))
+ .isInstanceOf(IllegalStateException.class);
Review comment:
```suggestion
final JobResultEntry otherEntryWithSameJobId = TestingJobResultStore.createJobResult(DUMMY_JOB_RESULT_ENTRY.getJobId(), ApplicationStatus.FAILED);
assertThatThrownBy(() -> jobResultStore.createDirtyResult(otherEntryWithSameJobId))
.isInstanceOf(IllegalStateException.class);
```
That way, we make it more explicit that we're actually checking on the JobID and not only on the entire instance (the test method name need to be aligned as well).
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/JobResultStoreContractTest.java
##########
@@ -0,0 +1,163 @@
+/*
+ * 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.runtime.highavailability;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.jobmaster.JobResult;
+import org.apache.flink.runtime.testutils.TestingJobResultStore;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.NoSuchElementException;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
+
+/**
+ * This interface defines a series of tests for any implementation of the {@link JobResultStore} to
+ * determine whether they correctly implement the contracts defined by the interface.
+ */
+public interface JobResultStoreContractTest {
+
+ JobResultEntry DUMMY_JOB_RESULT_ENTRY =
+ new JobResultEntry(TestingJobResultStore.DUMMY_JOB_RESULT);
+
+ JobResultStore createJobResultStore() throws IOException;
+
+ @Test
+ default void testStoreDirtyJobResult() throws IOException {
+ JobResultStore jobResultStore = createJobResultStore();
+ jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY);
+ assertThat(jobResultStore.hasDirtyJobResultEntry(DUMMY_JOB_RESULT_ENTRY.getJobId()))
+ .isTrue();
+ }
+
+ @Test
+ default void testStoreDuplicateJobResultThrowsException() throws IOException {
+ JobResultStore jobResultStore = createJobResultStore();
+ jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY);
+ assertThatThrownBy(() -> jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY))
+ .isInstanceOf(IllegalStateException.class);
+ }
+
+ @Test
+ default void testStoreCleanedJobResultThrowsException() throws IOException {
+ JobResultStore jobResultStore = createJobResultStore();
+ jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY);
+ jobResultStore.markResultAsClean(DUMMY_JOB_RESULT_ENTRY.getJobId());
+ assertThatThrownBy(() -> jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY))
+ .isInstanceOf(IllegalStateException.class);
+ }
+
+ @Test
+ default void testMarkDirtyResultAsClean() throws IOException {
+ JobResultStore jobResultStore = createJobResultStore();
+ jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY);
+ jobResultStore.markResultAsClean(DUMMY_JOB_RESULT_ENTRY.getJobId());
+ assertThat(jobResultStore.hasCleanJobResultEntry(DUMMY_JOB_RESULT_ENTRY.getJobId()))
+ .isTrue();
+ }
+
+ @Test
+ default void testCleaningDuplicateEntryThrowsNoException() throws IOException {
+ JobResultStore jobResultStore = createJobResultStore();
+ jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY);
+ jobResultStore.markResultAsClean(DUMMY_JOB_RESULT_ENTRY.getJobId());
+ assertThatNoException()
+ .isThrownBy(
+ () -> jobResultStore.markResultAsClean(DUMMY_JOB_RESULT_ENTRY.getJobId()));
+ }
+
+ @Test
+ default void testCleaningNonExistentEntryThrowsException() throws IOException {
+ JobResultStore jobResultStore = createJobResultStore();
+ assertThatThrownBy(
+ () -> jobResultStore.markResultAsClean(DUMMY_JOB_RESULT_ENTRY.getJobId()))
+ .isInstanceOf(NoSuchElementException.class);
+ }
+
+ @Test
+ default void testHasJobResultEntryWithDirtyEntry() throws IOException {
+ JobResultStore jobResultStore = createJobResultStore();
+ jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY);
+ assertThat(jobResultStore.hasDirtyJobResultEntry(DUMMY_JOB_RESULT_ENTRY.getJobId()))
+ .isTrue();
+ assertThat(jobResultStore.hasCleanJobResultEntry(DUMMY_JOB_RESULT_ENTRY.getJobId()))
+ .isFalse();
+ assertThat(jobResultStore.hasJobResultEntry(DUMMY_JOB_RESULT_ENTRY.getJobId())).isTrue();
+ }
+
+ @Test
+ default void testHasJobResultEntryWithCleanEntry() throws IOException {
+ JobResultStore jobResultStore = createJobResultStore();
+ jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY);
+ jobResultStore.markResultAsClean(DUMMY_JOB_RESULT_ENTRY.getJobId());
+ assertThat(jobResultStore.hasDirtyJobResultEntry(DUMMY_JOB_RESULT_ENTRY.getJobId()))
+ .isFalse();
+ assertThat(jobResultStore.hasCleanJobResultEntry(DUMMY_JOB_RESULT_ENTRY.getJobId()))
+ .isTrue();
+ assertThat(jobResultStore.hasJobResultEntry(DUMMY_JOB_RESULT_ENTRY.getJobId())).isTrue();
+ }
+
+ @Test
+ default void testHasJobResultEntryWithEmptyStore() throws IOException {
+ JobResultStore jobResultStore = createJobResultStore();
+ JobID jobId = new JobID();
+ assertThat(jobResultStore.hasDirtyJobResultEntry(jobId)).isFalse();
+ assertThat(jobResultStore.hasCleanJobResultEntry(jobId)).isFalse();
+ assertThat(jobResultStore.hasJobResultEntry(jobId)).isFalse();
+ }
+
+ @Test
+ default void testGetDirtyResultsWithNoEntry() throws IOException {
+ JobResultStore jobResultStore = createJobResultStore();
+ assertThat(jobResultStore.getDirtyResults()).isEmpty();
+ }
+
+ @Test
+ default void testGetDirtyResultsWithDirtyEntry() throws IOException {
+ JobResultStore jobResultStore = createJobResultStore();
+ jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY);
+ assertThat(
+ jobResultStore.getDirtyResults().stream()
+ .map(JobResult::getJobId)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(DUMMY_JOB_RESULT_ENTRY.getJobId());
+ }
+
+ @Test
+ default void testGetDirtyResultsWithDirtyAndCleanEntry() throws IOException {
+ JobResultStore jobResultStore = createJobResultStore();
+ jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY);
+ jobResultStore.markResultAsClean(DUMMY_JOB_RESULT_ENTRY.getJobId());
+
+ final JobResultEntry otherDirtyJobResultEntry =
+ new JobResultEntry(TestingJobResultStore.createSuccessfulJobResult(new JobID()));
+ jobResultStore.createDirtyResult(otherDirtyJobResultEntry);
+
+ assertThat(
+ jobResultStore.getDirtyResults().stream()
+ .map(JobResult::getJobId)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(otherDirtyJobResultEntry.getJobId());
Review comment:
```suggestion
assertThat(jobResultStore.getDirtyResults())
.singleElement()
.isEqualTo(otherDirtyJobResultEntry.getJobId());
```
It does the same. Feel free to use whichever version you like. I just wanted to show you `.singleElement()` assertj method. I just learned about it recently and find it quite handy in situations like that. that would apply to the test method above as well.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/AbstractThreadsafeJobResultStore.java
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.runtime.highavailability;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.jobmaster.JobResult;
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.IOException;
+import java.util.NoSuchElementException;
+import java.util.Set;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/** An abstract class for threadsafe implementations of the {@link JobResultStore}. */
+public abstract class AbstractThreadsafeJobResultStore implements JobResultStore {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(AbstractThreadsafeJobResultStore.class);
+
+ private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
+
+ @Override
+ public void createDirtyResult(JobResultEntry jobResultEntry) throws IOException {
+ Preconditions.checkState(
+ !hasJobResultEntry(jobResultEntry.getJobId()),
+ "Job result store already contains an entry for for job %s",
+ jobResultEntry.getJobId());
+
+ readWriteLock.writeLock().lock();
+ try {
+ createDirtyResultInternal(jobResultEntry);
+ } finally {
+ readWriteLock.writeLock().unlock();
+ }
+ }
+
+ @GuardedBy("readWriteLock")
+ protected abstract void createDirtyResultInternal(JobResultEntry jobResultEntry)
+ throws IOException;
+
+ @Override
+ public void markResultAsClean(JobID jobId) throws IOException, NoSuchElementException {
+ if (hasCleanJobResultEntry(jobId)) {
+ LOG.debug("The job {} is already marked as clean. No action required.", jobId);
+ return;
+ }
+
+ readWriteLock.writeLock().lock();
+ try {
+ markResultAsCleanInternal(jobId);
+ } finally {
+ readWriteLock.writeLock().unlock();
+ }
+ }
+
+ @GuardedBy("readWriteLock")
+ protected abstract void markResultAsCleanInternal(JobID jobId)
+ throws IOException, NoSuchElementException;
+
+ @Override
+ public boolean hasJobResultEntry(JobID jobId) throws IOException {
+ readWriteLock.readLock().lock();
+ try {
+ return hasDirtyJobResultEntryInternal(jobId) || hasCleanJobResultEntryInternal(jobId);
+ } finally {
+ readWriteLock.readLock().unlock();
+ }
+ }
+
+ @Override
+ public boolean hasDirtyJobResultEntry(JobID jobId) throws IOException {
+ readWriteLock.readLock().lock();
+ try {
+ return hasDirtyJobResultEntryInternal(jobId);
+ } finally {
+ readWriteLock.readLock().unlock();
+ }
+ }
+
+ @GuardedBy("readWriteLock")
+ protected abstract boolean hasDirtyJobResultEntryInternal(JobID jobId) throws IOException;
+
+ @Override
+ public boolean hasCleanJobResultEntry(JobID jobId) throws IOException {
+ readWriteLock.readLock().lock();
+ try {
+ return hasCleanJobResultEntryInternal(jobId);
+ } finally {
+ readWriteLock.readLock().unlock();
+ }
+ }
+
+ @GuardedBy("readWriteLock")
+ protected abstract boolean hasCleanJobResultEntryInternal(JobID jobId) throws IOException;
+
+ @Override
+ public Set<JobResult> getDirtyResults() throws IOException {
+ readWriteLock.readLock().lock();
+ try {
+ return getDirtyResultsInternal();
+ } finally {
+ readWriteLock.readLock().unlock();
+ }
+ }
+
+ @GuardedBy("readWriteLock")
+ protected abstract Set<JobResult> getDirtyResultsInternal() throws IOException;
Review comment:
That allows something like
```
@Override
public boolean hasJobResultEntry(JobID jobId) throws IOException {
return withReadLock(
() ->
hasDirtyJobResultEntryInternal(jobId)
|| hasCleanJobResultEntryInternal(jobId));
}
```
with no additional try catch being necessary...
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/FileSystemJobResultStoreTestInternal.java
##########
@@ -0,0 +1,196 @@
+/*
+ * 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.runtime.highavailability;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.jobmaster.JobResult;
+import org.apache.flink.runtime.testutils.TestingJobResultStore;
+import org.apache.flink.util.FileUtils;
+import org.apache.flink.util.TestLoggerExtension;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for the internal {@link FileSystemJobResultStore} mechanisms. */
+@ExtendWith(TestLoggerExtension.class)
+public class FileSystemJobResultStoreTestInternal {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private static final JobResultEntry DUMMY_JOB_RESULT_ENTRY =
Review comment:
Same here, we could reuse `JobResultStoreContractTest.DUMMY_JOB_RESULT_ENTRY`. WDYT?
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedJobResultStore.java
##########
@@ -20,112 +20,57 @@
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.highavailability.AbstractThreadsafeJobResultStore;
import org.apache.flink.runtime.highavailability.JobResultEntry;
import org.apache.flink.runtime.highavailability.JobResultStore;
import org.apache.flink.runtime.jobmaster.JobResult;
-import org.apache.flink.util.Preconditions;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.annotation.concurrent.GuardedBy;
+import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Set;
-import java.util.concurrent.locks.ReadWriteLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-import java.util.function.Supplier;
import java.util.stream.Collectors;
/** A thread-safe in-memory implementation of the {@link JobResultStore}. */
-public class EmbeddedJobResultStore implements JobResultStore {
-
- private static final Logger LOG = LoggerFactory.getLogger(EmbeddedJobResultStore.class);
-
- private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
+public class EmbeddedJobResultStore extends AbstractThreadsafeJobResultStore {
- @GuardedBy("readWriteLock")
- @VisibleForTesting
- final Map<JobID, JobResultEntry> dirtyJobResults = new HashMap<>();
+ @VisibleForTesting final Map<JobID, JobResultEntry> dirtyJobResults = new HashMap<>();
- @GuardedBy("readWriteLock")
- @VisibleForTesting
- final Map<JobID, JobResultEntry> cleanJobResults = new HashMap<>();
+ @VisibleForTesting final Map<JobID, JobResultEntry> cleanJobResults = new HashMap<>();
Review comment:
See my comment in `EmbeddedJobResultStoreInternal`. I guess, we can switch to `private` here without the `@VisibleForTesting` annotation.
--
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] autophagy commented on a change in pull request #18543: [FLINK-25431] Implement a file-based JobResultStore
Posted by GitBox <gi...@apache.org>.
autophagy commented on a change in pull request #18543:
URL: https://github.com/apache/flink/pull/18543#discussion_r794351052
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedJobResultStoreTestInternal.java
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.runtime.highavailability.nonha.embedded;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.highavailability.JobResultEntry;
+import org.apache.flink.runtime.testutils.TestingJobResultStore;
+import org.apache.flink.util.TestLoggerExtension;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.AssertionsForClassTypes.entry;
+
+/** Tests for the internal {@link EmbeddedJobResultStore} mechanisms. */
+@ExtendWith(TestLoggerExtension.class)
+public class EmbeddedJobResultStoreTestInternal {
Review comment:
Fair! That 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] flinkbot edited a comment on pull request #18543: [FLINK-25431] Implement a file-based JobResultStore
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18543:
URL: https://github.com/apache/flink/pull/18543#issuecomment-1023364026
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2cfc588752f0396b9c2d862585da6d7c2c46fb38",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30329",
"triggerID" : "2cfc588752f0396b9c2d862585da6d7c2c46fb38",
"triggerType" : "PUSH"
}, {
"hash" : "0d2150b89e191ae31e3bca29aed90d0508bc002a",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30380",
"triggerID" : "0d2150b89e191ae31e3bca29aed90d0508bc002a",
"triggerType" : "PUSH"
}, {
"hash" : "6e5a51ffbc7bbe8145a6f78d2e0c4f49bdcdc6ea",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30410",
"triggerID" : "6e5a51ffbc7bbe8145a6f78d2e0c4f49bdcdc6ea",
"triggerType" : "PUSH"
}, {
"hash" : "f62cd51ea1ab1d39834611586a7509fc0ecab7c0",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30425",
"triggerID" : "f62cd51ea1ab1d39834611586a7509fc0ecab7c0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0d2150b89e191ae31e3bca29aed90d0508bc002a Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30380)
* 6e5a51ffbc7bbe8145a6f78d2e0c4f49bdcdc6ea Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30410)
* f62cd51ea1ab1d39834611586a7509fc0ecab7c0 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30425)
<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] flinkbot edited a comment on pull request #18543: [FLINK-25431] Implement a file-based JobResultStore
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18543:
URL: https://github.com/apache/flink/pull/18543#issuecomment-1023364026
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2cfc588752f0396b9c2d862585da6d7c2c46fb38",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30329",
"triggerID" : "2cfc588752f0396b9c2d862585da6d7c2c46fb38",
"triggerType" : "PUSH"
}, {
"hash" : "0d2150b89e191ae31e3bca29aed90d0508bc002a",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30380",
"triggerID" : "0d2150b89e191ae31e3bca29aed90d0508bc002a",
"triggerType" : "PUSH"
}, {
"hash" : "6e5a51ffbc7bbe8145a6f78d2e0c4f49bdcdc6ea",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30410",
"triggerID" : "6e5a51ffbc7bbe8145a6f78d2e0c4f49bdcdc6ea",
"triggerType" : "PUSH"
}, {
"hash" : "f62cd51ea1ab1d39834611586a7509fc0ecab7c0",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "f62cd51ea1ab1d39834611586a7509fc0ecab7c0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0d2150b89e191ae31e3bca29aed90d0508bc002a Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30380)
* 6e5a51ffbc7bbe8145a6f78d2e0c4f49bdcdc6ea Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30410)
* f62cd51ea1ab1d39834611586a7509fc0ecab7c0 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] flinkbot edited a comment on pull request #18543: [FLINK-25431] Implement a file-based JobResultStore
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18543:
URL: https://github.com/apache/flink/pull/18543#issuecomment-1023364026
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2cfc588752f0396b9c2d862585da6d7c2c46fb38",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30329",
"triggerID" : "2cfc588752f0396b9c2d862585da6d7c2c46fb38",
"triggerType" : "PUSH"
}, {
"hash" : "0d2150b89e191ae31e3bca29aed90d0508bc002a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30380",
"triggerID" : "0d2150b89e191ae31e3bca29aed90d0508bc002a",
"triggerType" : "PUSH"
}, {
"hash" : "6e5a51ffbc7bbe8145a6f78d2e0c4f49bdcdc6ea",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30410",
"triggerID" : "6e5a51ffbc7bbe8145a6f78d2e0c4f49bdcdc6ea",
"triggerType" : "PUSH"
}, {
"hash" : "f62cd51ea1ab1d39834611586a7509fc0ecab7c0",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30425",
"triggerID" : "f62cd51ea1ab1d39834611586a7509fc0ecab7c0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6e5a51ffbc7bbe8145a6f78d2e0c4f49bdcdc6ea Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30410)
* f62cd51ea1ab1d39834611586a7509fc0ecab7c0 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30425)
<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] autophagy commented on a change in pull request #18543: [FLINK-25431] Implement a file-based JobResultStore
Posted by GitBox <gi...@apache.org>.
autophagy commented on a change in pull request #18543:
URL: https://github.com/apache/flink/pull/18543#discussion_r794350322
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/JobResultStoreContractTest.java
##########
@@ -0,0 +1,163 @@
+/*
+ * 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.runtime.highavailability;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.jobmaster.JobResult;
+import org.apache.flink.runtime.testutils.TestingJobResultStore;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.NoSuchElementException;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
+
+/**
+ * This interface defines a series of tests for any implementation of the {@link JobResultStore} to
+ * determine whether they correctly implement the contracts defined by the interface.
+ */
+public interface JobResultStoreContractTest {
+
+ JobResultEntry DUMMY_JOB_RESULT_ENTRY =
+ new JobResultEntry(TestingJobResultStore.DUMMY_JOB_RESULT);
+
+ JobResultStore createJobResultStore() throws IOException;
+
+ @Test
+ default void testStoreDirtyJobResult() throws IOException {
+ JobResultStore jobResultStore = createJobResultStore();
+ jobResultStore.createDirtyResult(DUMMY_JOB_RESULT_ENTRY);
+ assertThat(jobResultStore.hasDirtyJobResultEntry(DUMMY_JOB_RESULT_ENTRY.getJobId()))
+ .isTrue();
+ }
Review comment:
That's a good point. I wrote the contract tests by, more-or-less, reading over the javadocs for the interface and creating a test for everything mentioned. But you're right, testing the presence of entrys in the store necessarily tests the `hasEntry` functions in any case, so the earlier tests are a bit superfluous.
--
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 edited a comment on pull request #18543: [FLINK-25431] Implement a file-based JobResultStore
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18543:
URL: https://github.com/apache/flink/pull/18543#issuecomment-1023364026
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2cfc588752f0396b9c2d862585da6d7c2c46fb38",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30329",
"triggerID" : "2cfc588752f0396b9c2d862585da6d7c2c46fb38",
"triggerType" : "PUSH"
}, {
"hash" : "0d2150b89e191ae31e3bca29aed90d0508bc002a",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30380",
"triggerID" : "0d2150b89e191ae31e3bca29aed90d0508bc002a",
"triggerType" : "PUSH"
}, {
"hash" : "6e5a51ffbc7bbe8145a6f78d2e0c4f49bdcdc6ea",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30410",
"triggerID" : "6e5a51ffbc7bbe8145a6f78d2e0c4f49bdcdc6ea",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2cfc588752f0396b9c2d862585da6d7c2c46fb38 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30329)
* 0d2150b89e191ae31e3bca29aed90d0508bc002a Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30380)
* 6e5a51ffbc7bbe8145a6f78d2e0c4f49bdcdc6ea Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30410)
<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] flinkbot edited a comment on pull request #18543: [FLINK-25431] Implement a file-based JobResultStore
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18543:
URL: https://github.com/apache/flink/pull/18543#issuecomment-1023364026
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2cfc588752f0396b9c2d862585da6d7c2c46fb38",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30329",
"triggerID" : "2cfc588752f0396b9c2d862585da6d7c2c46fb38",
"triggerType" : "PUSH"
}, {
"hash" : "0d2150b89e191ae31e3bca29aed90d0508bc002a",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30380",
"triggerID" : "0d2150b89e191ae31e3bca29aed90d0508bc002a",
"triggerType" : "PUSH"
}, {
"hash" : "6e5a51ffbc7bbe8145a6f78d2e0c4f49bdcdc6ea",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30410",
"triggerID" : "6e5a51ffbc7bbe8145a6f78d2e0c4f49bdcdc6ea",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0d2150b89e191ae31e3bca29aed90d0508bc002a Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30380)
* 6e5a51ffbc7bbe8145a6f78d2e0c4f49bdcdc6ea Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30410)
<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] flinkbot edited a comment on pull request #18543: [FLINK-25431] Implement a file-based JobResultStore
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18543:
URL: https://github.com/apache/flink/pull/18543#issuecomment-1023364026
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2cfc588752f0396b9c2d862585da6d7c2c46fb38",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30329",
"triggerID" : "2cfc588752f0396b9c2d862585da6d7c2c46fb38",
"triggerType" : "PUSH"
}, {
"hash" : "0d2150b89e191ae31e3bca29aed90d0508bc002a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30380",
"triggerID" : "0d2150b89e191ae31e3bca29aed90d0508bc002a",
"triggerType" : "PUSH"
}, {
"hash" : "6e5a51ffbc7bbe8145a6f78d2e0c4f49bdcdc6ea",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30410",
"triggerID" : "6e5a51ffbc7bbe8145a6f78d2e0c4f49bdcdc6ea",
"triggerType" : "PUSH"
}, {
"hash" : "f62cd51ea1ab1d39834611586a7509fc0ecab7c0",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30425",
"triggerID" : "f62cd51ea1ab1d39834611586a7509fc0ecab7c0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f62cd51ea1ab1d39834611586a7509fc0ecab7c0 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30425)
<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] autophagy commented on a change in pull request #18543: [FLINK-25431] Implement a file-based JobResultStore
Posted by GitBox <gi...@apache.org>.
autophagy commented on a change in pull request #18543:
URL: https://github.com/apache/flink/pull/18543#discussion_r794339834
##########
File path: docs/layouts/shortcodes/generated/common_high_availability_jrs_section.html
##########
@@ -0,0 +1,24 @@
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Key</th>
+ <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>job-result-store.delete-on-commit</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Determines whether job results should be automatically removed from the underlying job result store when they are clean. If false, the cleaned job results are, instead, marked as clean to indicate their state. In this case, Flink no longer hasownership and the resources need to be cleaned up by theuser.</td>
+ </tr>
+ <tr>
+ <td><h5>job-result-store.storage-path</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Defines where job results should be stored. This should be an underlying file-system that provides read-after-write consistency. By default, this is <code class="highlighter-rouge">high-availability.storageDir/job-result-store/high-availability.cluster-id</code>.</td>
Review comment:
Yeah, this is much clearer in the resulting documentation. I'll change this as suggested in the Option itself, but I think it might be a worthwhile extension to the `org.apache.flink.configuration.description.DescriptionBuilder`.
--
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 edited a comment on pull request #18543: [FLINK-25431] Implement a file-based JobResultStore
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18543:
URL: https://github.com/apache/flink/pull/18543#issuecomment-1023364026
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2cfc588752f0396b9c2d862585da6d7c2c46fb38",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30329",
"triggerID" : "2cfc588752f0396b9c2d862585da6d7c2c46fb38",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2cfc588752f0396b9c2d862585da6d7c2c46fb38 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30329)
<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] flinkbot edited a comment on pull request #18543: [FLINK-25431] Implement a file-based JobResultStore
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18543:
URL: https://github.com/apache/flink/pull/18543#issuecomment-1023364026
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2cfc588752f0396b9c2d862585da6d7c2c46fb38",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30329",
"triggerID" : "2cfc588752f0396b9c2d862585da6d7c2c46fb38",
"triggerType" : "PUSH"
}, {
"hash" : "0d2150b89e191ae31e3bca29aed90d0508bc002a",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30380",
"triggerID" : "0d2150b89e191ae31e3bca29aed90d0508bc002a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2cfc588752f0396b9c2d862585da6d7c2c46fb38 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30329)
* 0d2150b89e191ae31e3bca29aed90d0508bc002a Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30380)
<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] autophagy commented on a change in pull request #18543: [FLINK-25431] Implement a file-based JobResultStore
Posted by GitBox <gi...@apache.org>.
autophagy commented on a change in pull request #18543:
URL: https://github.com/apache/flink/pull/18543#discussion_r794307910
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/AbstractThreadsafeJobResultStore.java
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.runtime.highavailability;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.jobmaster.JobResult;
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.IOException;
+import java.util.NoSuchElementException;
+import java.util.Set;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/** An abstract class for threadsafe implementations of the {@link JobResultStore}. */
+public abstract class AbstractThreadsafeJobResultStore implements JobResultStore {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(AbstractThreadsafeJobResultStore.class);
+
+ private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
+
+ @Override
+ public void createDirtyResult(JobResultEntry jobResultEntry) throws IOException {
+ Preconditions.checkState(
+ !hasJobResultEntry(jobResultEntry.getJobId()),
+ "Job result store already contains an entry for for job %s",
+ jobResultEntry.getJobId());
+
+ readWriteLock.writeLock().lock();
+ try {
+ createDirtyResultInternal(jobResultEntry);
+ } finally {
+ readWriteLock.writeLock().unlock();
+ }
+ }
+
+ @GuardedBy("readWriteLock")
+ protected abstract void createDirtyResultInternal(JobResultEntry jobResultEntry)
+ throws IOException;
+
+ @Override
+ public void markResultAsClean(JobID jobId) throws IOException, NoSuchElementException {
+ if (hasCleanJobResultEntry(jobId)) {
+ LOG.debug("The job {} is already marked as clean. No action required.", jobId);
+ return;
+ }
+
+ readWriteLock.writeLock().lock();
+ try {
+ markResultAsCleanInternal(jobId);
+ } finally {
+ readWriteLock.writeLock().unlock();
+ }
+ }
+
+ @GuardedBy("readWriteLock")
+ protected abstract void markResultAsCleanInternal(JobID jobId)
+ throws IOException, NoSuchElementException;
+
+ @Override
+ public boolean hasJobResultEntry(JobID jobId) throws IOException {
+ readWriteLock.readLock().lock();
+ try {
+ return hasDirtyJobResultEntryInternal(jobId) || hasCleanJobResultEntryInternal(jobId);
+ } finally {
+ readWriteLock.readLock().unlock();
+ }
+ }
+
+ @Override
+ public boolean hasDirtyJobResultEntry(JobID jobId) throws IOException {
+ readWriteLock.readLock().lock();
+ try {
+ return hasDirtyJobResultEntryInternal(jobId);
+ } finally {
+ readWriteLock.readLock().unlock();
+ }
+ }
+
+ @GuardedBy("readWriteLock")
+ protected abstract boolean hasDirtyJobResultEntryInternal(JobID jobId) throws IOException;
+
+ @Override
+ public boolean hasCleanJobResultEntry(JobID jobId) throws IOException {
+ readWriteLock.readLock().lock();
+ try {
+ return hasCleanJobResultEntryInternal(jobId);
+ } finally {
+ readWriteLock.readLock().unlock();
+ }
+ }
+
+ @GuardedBy("readWriteLock")
+ protected abstract boolean hasCleanJobResultEntryInternal(JobID jobId) throws IOException;
+
+ @Override
+ public Set<JobResult> getDirtyResults() throws IOException {
+ readWriteLock.readLock().lock();
+ try {
+ return getDirtyResultsInternal();
+ } finally {
+ readWriteLock.readLock().unlock();
+ }
+ }
+
+ @GuardedBy("readWriteLock")
+ protected abstract Set<JobResult> getDirtyResultsInternal() throws IOException;
Review comment:
Ah wonderful, this is exactly what i meant! Thank you for the heads up
--
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 edited a comment on pull request #18543: [FLINK-25431] Implement a file-based JobResultStore
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18543:
URL: https://github.com/apache/flink/pull/18543#issuecomment-1023364026
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2cfc588752f0396b9c2d862585da6d7c2c46fb38",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30329",
"triggerID" : "2cfc588752f0396b9c2d862585da6d7c2c46fb38",
"triggerType" : "PUSH"
}, {
"hash" : "0d2150b89e191ae31e3bca29aed90d0508bc002a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "0d2150b89e191ae31e3bca29aed90d0508bc002a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2cfc588752f0396b9c2d862585da6d7c2c46fb38 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30329)
* 0d2150b89e191ae31e3bca29aed90d0508bc002a 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] flinkbot edited a comment on pull request #18543: [FLINK-25431] Implement a file-based JobResultStore
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18543:
URL: https://github.com/apache/flink/pull/18543#issuecomment-1023364026
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2cfc588752f0396b9c2d862585da6d7c2c46fb38",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30329",
"triggerID" : "2cfc588752f0396b9c2d862585da6d7c2c46fb38",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2cfc588752f0396b9c2d862585da6d7c2c46fb38 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30329)
<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] autophagy commented on a change in pull request #18543: [FLINK-25431] Implement a file-based JobResultStore
Posted by GitBox <gi...@apache.org>.
autophagy commented on a change in pull request #18543:
URL: https://github.com/apache/flink/pull/18543#discussion_r794350716
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedJobResultStoreTestInternal.java
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.runtime.highavailability.nonha.embedded;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.highavailability.JobResultEntry;
+import org.apache.flink.runtime.testutils.TestingJobResultStore;
+import org.apache.flink.util.TestLoggerExtension;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.AssertionsForClassTypes.entry;
+
+/** Tests for the internal {@link EmbeddedJobResultStore} mechanisms. */
+@ExtendWith(TestLoggerExtension.class)
+public class EmbeddedJobResultStoreTestInternal {
+
+ private static final JobResultEntry DUMMY_JOB_RESULT_ENTRY =
Review comment:
I think it already is public, so using it in the internal tests should be pretty straightforward.
--
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