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 2021/05/04 19:41:41 UTC
[GitHub] [flink] XComp opened a new pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
XComp opened a new pull request #15832:
URL: https://github.com/apache/flink/pull/15832
## What is the purpose of the change
Updating the Checkpoint Metadata store of ZooKeeper and Kubernetes might lead into situations where the data was actually written but the request still failed (due to network issue for instance). In these cases, we end up in an inconsistent state where a reference exists in ZooKeeper or the k8s ConfigMap but the corresponding Checkpoint's state is discarded due to the failure. That might cause problems during recovery.
## Brief change log
* Introduces `PossibleInconsistentStateException`
* `Fabric8FlinkKubeClient` translates network issues into `PossibleInconsistentStateException`
* `StateHandleStore` and its implementing classes handle and forward `PossibleInconsistentStateException`
* Extends `TestingLongStateHandleHelper` to work with reference instead of saving the state in the handle itself. The previous implementation caused problems when deserializing the StateHandles not referencing the actual data but providing a copy of the state.
* `CheckpointCoordinator` handles `PossibleInconsistentStateException`
## Verifying this change
The following tests were extended:
* `CheckpointCoordinatorFailureTest` checks the failure handling for `PossibleInconsistentStateException` and another exception
* `KubernetesStateHandleStoreTest` and `ZooKeeperStateHandleStoreTest` were extended to cover the relevant exceptions
* `Fabric8FlinkKubeClientTest` was extended to cover both failure cases
## 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/Mesos, ZooKeeper: yes
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
XComp commented on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-839879710
I finally rebase happened. I created the backports for `release-1.12` and `release-1.13`
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-832214416
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17556",
"triggerID" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"triggerType" : "PUSH"
}, {
"hash" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"triggerType" : "PUSH"
}, {
"hash" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17565",
"triggerID" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"triggerType" : "PUSH"
}, {
"hash" : "4b34da74d08c5777b4fe69060acfea6738425844",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4b34da74d08c5777b4fe69060acfea6738425844",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 707261cd4e5b59e7e73fcbd55d3f76c66aebec48 UNKNOWN
* 7fad07e7c67e05691fdf158173e84017c05cbeb0 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17565)
* 4b34da74d08c5777b4fe69060acfea6738425844 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a change in pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
XComp commented on a change in pull request #15832:
URL: https://github.com/apache/flink/pull/15832#discussion_r626296284
##########
File path: flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesStateHandleStoreITCase.java
##########
@@ -68,8 +68,8 @@ public void testMultipleKubernetesStateHandleStores() throws Exception {
new TestingLeaderCallbackHandler[leaderNum];
@SuppressWarnings("unchecked")
- final KubernetesStateHandleStore<Long>[] stateHandleStores =
- new KubernetesStateHandleStore[leaderNum];
+ final KubernetesStateHandleStore<TestingLongStateHandleHelper.LongStateHandle>[]
Review comment:
The Kubernetes-related ITCases are currently not running. I created [FLINK-20564](https://issues.apache.org/jira/browse/FLINK-22564) to cover this. I haven't had the chance to run the ITCase locally, yet.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-832214416
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17556",
"triggerID" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a14a487a208ab8c5ea047665e472fd27f829e51c Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17556)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-832214416
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17556",
"triggerID" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"triggerType" : "PUSH"
}, {
"hash" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"triggerType" : "PUSH"
}, {
"hash" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17565",
"triggerID" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"triggerType" : "PUSH"
}, {
"hash" : "4b34da74d08c5777b4fe69060acfea6738425844",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17854",
"triggerID" : "4b34da74d08c5777b4fe69060acfea6738425844",
"triggerType" : "PUSH"
}, {
"hash" : "44bb7657519abe9578d2b19ed65f668c0538c828",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44bb7657519abe9578d2b19ed65f668c0538c828",
"triggerType" : "PUSH"
}, {
"hash" : "13993329ba9c544413f04d23a8d0038dabc6b638",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17899",
"triggerID" : "13993329ba9c544413f04d23a8d0038dabc6b638",
"triggerType" : "PUSH"
}, {
"hash" : "c9288127b18f00887b82b710bd42fc1df1487753",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17920",
"triggerID" : "c9288127b18f00887b82b710bd42fc1df1487753",
"triggerType" : "PUSH"
}, {
"hash" : "4c7f59bea05d020743c4aa546e569729e2a7a92a",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17928",
"triggerID" : "4c7f59bea05d020743c4aa546e569729e2a7a92a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 707261cd4e5b59e7e73fcbd55d3f76c66aebec48 UNKNOWN
* 44bb7657519abe9578d2b19ed65f668c0538c828 UNKNOWN
* c9288127b18f00887b82b710bd42fc1df1487753 Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17920)
* 4c7f59bea05d020743c4aa546e569729e2a7a92a Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17928)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #15832:
URL: https://github.com/apache/flink/pull/15832#discussion_r630164743
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperStateHandleStore.java
##########
@@ -126,56 +144,71 @@ public ZooKeeperStateHandleStore(
* @param pathInZooKeeper Destination path in ZooKeeper (expected to *not* exist yet)
* @param state State to be added
* @return The Created {@link RetrievableStateHandle}.
+ * @throws PossibleInconsistentStateException if the write-to-ZooKeeper operation failed. This
+ * indicates that it's not clear whether the new state was successfully written to ZooKeeper
+ * or not. Proper error handling has to be applied on the caller's side.
* @throws Exception If a ZooKeeper or state handle operation fails
*/
@Override
- public RetrievableStateHandle<T> addAndLock(String pathInZooKeeper, T state) throws Exception {
+ public RetrievableStateHandle<T> addAndLock(String pathInZooKeeper, T state)
+ throws PossibleInconsistentStateException, Exception {
checkNotNull(pathInZooKeeper, "Path in ZooKeeper");
checkNotNull(state, "State");
final String path = normalizePath(pathInZooKeeper);
RetrievableStateHandle<T> storeHandle = storage.store(state);
- boolean success = false;
+ byte[] serializedStoreHandle = serializeStateHandle(storeHandle);
+ // initialize flag to serve the failure case
+ boolean discardState = true;
try {
- // Serialize the state handle. This writes the state to the backend.
- byte[] serializedStoreHandle = InstantiationUtil.serializeObject(storeHandle);
-
- // Write state handle (not the actual state) to ZooKeeper. This is expected to be
- // smaller than the state itself. This level of indirection makes sure that data in
- // ZooKeeper is small, because ZooKeeper is designed for data in the KB range, but
- // the state can be larger.
- // Create the lock node in a transaction with the actual state node. That way we can
- // prevent
- // race conditions with a concurrent delete operation.
- client.inTransaction()
- .create()
- .withMode(CreateMode.PERSISTENT)
- .forPath(path, serializedStoreHandle)
- .and()
- .create()
- .withMode(CreateMode.EPHEMERAL)
- .forPath(getLockPath(path))
- .and()
- .commit();
-
- success = true;
+ writeStoreHandleTransactionally(path, serializedStoreHandle);
+ discardState = false;
return storeHandle;
- } catch (KeeperException.NodeExistsException e) {
+ } catch (Exception e) {
+ if (indicatesPossiblyInconsistentState(e)) {
+ discardState = false;
+ throw new PossibleInconsistentStateException(e);
+ }
+
// We wrap the exception here so that it could be caught in DefaultJobGraphStore
- throw new AlreadyExistException("ZooKeeper node " + path + " already exists.", e);
+ throw ExceptionUtils.findThrowable(e, KeeperException.NodeExistsException.class)
Review comment:
Is this line reachable? In `indicatesPossiblyInconsistentState` above, we already checked for `NodeExistsException` and thrown an exception if it was found.
I think it's better to place this check before the `indicatesPossiblyInconsistentState` call (rather than updating the map).
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperStateHandleStore.java
##########
@@ -126,56 +144,71 @@ public ZooKeeperStateHandleStore(
* @param pathInZooKeeper Destination path in ZooKeeper (expected to *not* exist yet)
* @param state State to be added
* @return The Created {@link RetrievableStateHandle}.
+ * @throws PossibleInconsistentStateException if the write-to-ZooKeeper operation failed. This
+ * indicates that it's not clear whether the new state was successfully written to ZooKeeper
+ * or not. Proper error handling has to be applied on the caller's side.
* @throws Exception If a ZooKeeper or state handle operation fails
*/
@Override
- public RetrievableStateHandle<T> addAndLock(String pathInZooKeeper, T state) throws Exception {
+ public RetrievableStateHandle<T> addAndLock(String pathInZooKeeper, T state)
+ throws PossibleInconsistentStateException, Exception {
checkNotNull(pathInZooKeeper, "Path in ZooKeeper");
checkNotNull(state, "State");
final String path = normalizePath(pathInZooKeeper);
RetrievableStateHandle<T> storeHandle = storage.store(state);
- boolean success = false;
+ byte[] serializedStoreHandle = serializeStateHandle(storeHandle);
+ // initialize flag to serve the failure case
+ boolean discardState = true;
try {
- // Serialize the state handle. This writes the state to the backend.
- byte[] serializedStoreHandle = InstantiationUtil.serializeObject(storeHandle);
-
- // Write state handle (not the actual state) to ZooKeeper. This is expected to be
- // smaller than the state itself. This level of indirection makes sure that data in
- // ZooKeeper is small, because ZooKeeper is designed for data in the KB range, but
- // the state can be larger.
- // Create the lock node in a transaction with the actual state node. That way we can
- // prevent
- // race conditions with a concurrent delete operation.
- client.inTransaction()
- .create()
- .withMode(CreateMode.PERSISTENT)
- .forPath(path, serializedStoreHandle)
- .and()
- .create()
- .withMode(CreateMode.EPHEMERAL)
- .forPath(getLockPath(path))
- .and()
- .commit();
-
- success = true;
+ writeStoreHandleTransactionally(path, serializedStoreHandle);
+ discardState = false;
return storeHandle;
- } catch (KeeperException.NodeExistsException e) {
+ } catch (Exception e) {
+ if (indicatesPossiblyInconsistentState(e)) {
+ discardState = false;
+ throw new PossibleInconsistentStateException(e);
+ }
+
// We wrap the exception here so that it could be caught in DefaultJobGraphStore
- throw new AlreadyExistException("ZooKeeper node " + path + " already exists.", e);
+ throw ExceptionUtils.findThrowable(e, KeeperException.NodeExistsException.class)
+ .map(
+ nee ->
+ new AlreadyExistException(
+ "ZooKeeper node " + path + " already exists.", nee))
+ .orElseThrow(() -> e);
} finally {
- if (!success) {
- // Cleanup the state handle if it was not written to ZooKeeper.
- if (storeHandle != null) {
- storeHandle.discardState();
- }
+ if (discardState) {
+ storeHandle.discardState();
Review comment:
I think we could get rid of a boolean flag and simplify code a bit by discarding the handle in the `catch` block instead of `finally`, something like this:
```
try {
writeStoreHandleTransactionally();
} catch (Exception e) {
if (indicatesPossiblyInconsistentState) {
throw PossibleInconsistentStateException;
} else {
storeHandle.discardState();
}
}
```
WDYT?
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/util/StateHandleStoreUtils.java
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.util;
+
+import org.apache.flink.runtime.persistence.StateHandleStore;
+import org.apache.flink.runtime.state.StateObject;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.InstantiationUtil;
+
+/**
+ * {@code StateHandleStoreUtils} collects utility methods that might be usefule for any {@link
+ * StateHandleStore} implementation.
+ */
+public class StateHandleStoreUtils {
+
+ /**
+ * Serializes the passed {@link StateObject} and discards the state in case of failure.
+ *
+ * @param stateObject the {@code StateObject} that shall be serialized.
+ * @return The serialized version of the passed {@code StateObject}.
+ * @throws Exception if an error occurred during the serialization. The corresponding {@code
+ * StateObject} will be discarded in that case.
+ */
+ public static byte[] serializeStateHandle(StateObject stateObject) throws Exception {
Review comment:
I think it would be more consistent to have a symmetric method in this class to deserialize handle.
`ZooKeeperStateHandleStore` currently calls `InstantiationUtil.deserializeObject` for that.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/util/StateHandleStoreUtils.java
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.util;
+
+import org.apache.flink.runtime.persistence.StateHandleStore;
+import org.apache.flink.runtime.state.StateObject;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.InstantiationUtil;
+
+/**
+ * {@code StateHandleStoreUtils} collects utility methods that might be usefule for any {@link
+ * StateHandleStore} implementation.
+ */
+public class StateHandleStoreUtils {
+
+ /**
+ * Serializes the passed {@link StateObject} and discards the state in case of failure.
+ *
+ * @param stateObject the {@code StateObject} that shall be serialized.
+ * @return The serialized version of the passed {@code StateObject}.
+ * @throws Exception if an error occurred during the serialization. The corresponding {@code
+ * StateObject} will be discarded in that case.
+ */
+ public static byte[] serializeStateHandle(StateObject stateObject) throws Exception {
+ try {
+ return InstantiationUtil.serializeObject(stateObject);
+ } catch (Exception e) {
+ try {
+ stateObject.discardState();
Review comment:
**nit**: this contract isn't very obvious to me, maybe it makes sense to rename the method to `serializeOrDiscard` or similar
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperStateHandleStore.java
##########
@@ -196,29 +229,55 @@ public void replace(String pathInZooKeeper, IntegerResourceVersion expectedVersi
RetrievableStateHandle<T> newStateHandle = storage.store(state);
- boolean success = false;
+ final byte[] serializedStateHandle = serializeStateHandle(newStateHandle);
+ // initialize flags to serve the failure case
+ boolean discardOldState = false;
+ boolean discardNewState = true;
try {
- // Serialize the new state handle. This writes the state to the backend.
- byte[] serializedStateHandle = InstantiationUtil.serializeObject(newStateHandle);
+ setStateHandle(path, serializedStateHandle, expectedVersion.getValue());
+
+ // swap subject for deletion in case of success
+ discardOldState = true;
+ discardNewState = false;
+ } catch (Exception e) {
+ if (indicatesPossiblyInconsistentState(e)) {
+ // it's unclear whether the state handle metadata was written to ZooKeeper -
+ // hence, we don't discard any data
+ discardNewState = false;
+ throw new PossibleInconsistentStateException(e);
+ }
- // Replace state handle in ZooKeeper.
- client.setData()
- .withVersion(expectedVersion.getValue())
- .forPath(path, serializedStateHandle);
- success = true;
- } catch (KeeperException.NoNodeException e) {
// We wrap the exception here so that it could be caught in DefaultJobGraphStore
- throw new NotExistException("ZooKeeper node " + path + " does not exist.", e);
+ throw ExceptionUtils.findThrowable(e, KeeperException.NoNodeException.class)
+ .map(
+ nnee ->
+ new NotExistException(
+ "ZooKeeper node " + path + " does not exist.", nnee))
+ .orElseThrow(() -> e);
} finally {
- if (success) {
+ if (discardOldState) {
oldStateHandle.discardState();
- } else {
+ }
+
+ if (discardNewState) {
newStateHandle.discardState();
}
}
}
+ // this method is provided for the sole purpose of easier testing
+ @VisibleForTesting
+ protected void setStateHandle(String path, byte[] serializedStateHandle, int expectedVersion)
+ throws Exception {
+ // Replace state handle in ZooKeeper.
+ client.setData().withVersion(expectedVersion).forPath(path, serializedStateHandle);
+ }
+
+ private boolean indicatesPossiblyInconsistentState(Exception e) {
+ return !PRE_COMMIT_EXCEPTIONS.contains(e.getClass());
Review comment:
Checking for class means that no subclass will match. Maybe `NoAuthException` extended by `AuthExpiredException`.
I'm not sure whether this is a real issue though.
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorFailureTest.java
##########
@@ -160,8 +177,94 @@ public void testFailingCompletedCheckpointStoreAdd() throws Exception {
.discardState();
}
+ @Test
+ public void testCleanupForGenericFailure() throws Exception {
+ testStoringFailureHandling(new FlinkRuntimeException("Expected exception"), 1);
+ }
+
+ @Test
+ public void testCleanupOmissionForPossibleInconsistentStateException() throws Exception {
+ testStoringFailureHandling(new PossibleInconsistentStateException(), 0);
+ }
+
+ private void testStoringFailureHandling(Exception failure, int expectedCleanupCalls)
+ throws Exception {
+ final JobVertexID jobVertexID1 = new JobVertexID();
+
+ final ExecutionGraph graph =
+ new CheckpointCoordinatorTestingUtils.CheckpointExecutionGraphBuilder()
+ .addJobVertex(jobVertexID1)
+ .build();
+
+ final ExecutionVertex vertex = graph.getJobVertex(jobVertexID1).getTaskVertices()[0];
+ final ExecutionAttemptID attemptId = vertex.getCurrentExecutionAttempt().getAttemptId();
+
+ final StandaloneCheckpointIDCounter checkpointIDCounter =
+ new StandaloneCheckpointIDCounter();
+
+ final ManuallyTriggeredScheduledExecutor manuallyTriggeredScheduledExecutor =
+ new ManuallyTriggeredScheduledExecutor();
+
+ final CompletedCheckpointStore completedCheckpointStore =
+ new FailingCompletedCheckpointStore(
+ (checkpoint, ignoredCleaner, ignoredPostCleanCallback) -> {
+ throw failure;
+ });
+
+ final AtomicInteger cleanupCallCount = new AtomicInteger(0);
+ final CheckpointCoordinator checkpointCoordinator =
+ new CheckpointCoordinatorBuilder()
+ .setExecutionGraph(graph)
+ .setCheckpointIDCounter(checkpointIDCounter)
+ .setCheckpointsCleaner(
+ new CheckpointsCleaner() {
+
+ private static final long serialVersionUID =
+ 2029876992397573325L;
+
+ @Override
+ public void cleanCheckpointOnFailedStoring(
+ CompletedCheckpoint completedCheckpoint,
+ Executor executor) {
+ cleanupCallCount.incrementAndGet();
+ super.cleanCheckpointOnFailedStoring(
+ completedCheckpoint, executor);
+ }
+ })
+ .setCompletedCheckpointStore(completedCheckpointStore)
+ .setTimer(manuallyTriggeredScheduledExecutor)
+ .build();
+ checkpointCoordinator.triggerSavepoint(tmpFolder.newFolder().getAbsolutePath());
+ manuallyTriggeredScheduledExecutor.triggerAll();
+
+ try {
+ checkpointCoordinator.receiveAcknowledgeMessage(
+ new AcknowledgeCheckpoint(
+ graph.getJobID(), attemptId, checkpointIDCounter.getLast()),
+ "unknown location");
+ fail("CheckpointException should have been thrown.");
+ } catch (CheckpointException e) {
+ assertThat(
+ e.getCheckpointFailureReason(),
+ is(CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE));
+ }
+
+ assertThat(cleanupCallCount.get(), is(expectedCleanupCalls));
+ }
+
private static final class FailingCompletedCheckpointStore implements CompletedCheckpointStore {
+ private final TriConsumerWithException<
+ CompletedCheckpoint, CheckpointsCleaner, Runnable, Exception>
+ addCheckpointConsumer;
+
+ public FailingCompletedCheckpointStore(
+ TriConsumerWithException<
+ CompletedCheckpoint, CheckpointsCleaner, Runnable, Exception>
+ addCheckpointConsumer) {
+ this.addCheckpointConsumer = addCheckpointConsumer;
Review comment:
I'm not sure whether the flexibility we get here with TriConsumer justifies the complexity added.
Wouldn't a field `private Exception failure;` passed to constructor sufficicient?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] tillrohrmann commented on a change in pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #15832:
URL: https://github.com/apache/flink/pull/15832#discussion_r628363445
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/persistence/PossibleInconsistentStateException.java
##########
@@ -0,0 +1,36 @@
+/*
+ * 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.persistence;
+
+/**
+ * {@code PossibleInconsistentStateException} represents errors that might have lead to an
+ * inconsistent state within the HA resources.
+ */
+public class PossibleInconsistentStateException extends Exception {
Review comment:
Shall we let this exception extend from `FlinkException`?
##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesStateHandleStore.java
##########
@@ -114,21 +118,28 @@ public KubernetesStateHandleStore(
* @param key Key in ConfigMap
* @param state State to be added
* @throws AlreadyExistException if the name already exists
+ * @throws PossibleInconsistentStateException if the write-to-Kubernetes operation failed. This
+ * indicates that it's not clear whether the new state was successfully written to
+ * Kubernetes or not. No state was discarded. Proper error handling has to be applied on the
+ * caller's side.
* @throws Exception if persisting state or writing state handle failed
*/
@Override
- public RetrievableStateHandle<T> addAndLock(String key, T state) throws Exception {
+ public RetrievableStateHandle<T> addAndLock(String key, T state)
+ throws PossibleInconsistentStateException, Exception {
checkNotNull(key, "Key in ConfigMap.");
checkNotNull(state, "State.");
final RetrievableStateHandle<T> storeHandle = storage.store(state);
- boolean success = false;
+ final byte[] serializedStoreHandle = serializeStateHandle(storeHandle);
+ // initialize flag to serve the failure case
+ boolean discardState = true;
try {
- final byte[] serializedStoreHandle = InstantiationUtil.serializeObject(storeHandle);
- success =
- kubeClient
+ // a successful operation will result in the state not being discarded
+ discardState =
+ !kubeClient
Review comment:
I think this is personal taste: I find it easier to think about it with `success` and w/o the negation here.
##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesStateHandleStore.java
##########
@@ -470,6 +508,26 @@ public String toString() {
return this.getClass().getSimpleName() + "{configMapName='" + configMapName + "'}";
}
+ private byte[] serializeStateHandle(RetrievableStateHandle<T> retrievableStateHandle)
+ throws Exception {
+ try {
+ // Serialize the state handle. This writes the state to the backend.
+ return InstantiationUtil.serializeObject(retrievableStateHandle);
+ } catch (Exception e) {
+ discardState(retrievableStateHandle);
Review comment:
I think here it would be good to catch a possible exception and add it as a suppressed exception to `e`. Otherwise, we might lose it.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperStateHandleStore.java
##########
@@ -196,27 +231,74 @@ public void replace(String pathInZooKeeper, IntegerResourceVersion expectedVersi
RetrievableStateHandle<T> newStateHandle = storage.store(state);
- boolean success = false;
+ final byte[] serializedStateHandle = serializeStateHandle(newStateHandle);
+ // initialize flags to serve the failure case
+ boolean discardOldState = false;
+ boolean discardNewState = true;
try {
- // Serialize the new state handle. This writes the state to the backend.
- byte[] serializedStateHandle = InstantiationUtil.serializeObject(newStateHandle);
+ setStateHandle(path, serializedStateHandle, expectedVersion.getValue());
+
+ // swap subject for deletion in case of success
+ discardOldState = true;
+ discardNewState = false;
+ } catch (Exception e) {
+ if (indicatesPossiblyInconsistentState(e)) {
+ // it's unclear whether the state handle metadata was written to ZooKeeper -
+ // hence, we don't discard any data
+ discardOldState = false;
+ discardNewState = false;
+ throw new PossibleInconsistentStateException(e);
+ }
- // Replace state handle in ZooKeeper.
- client.setData()
- .withVersion(expectedVersion.getValue())
- .forPath(path, serializedStateHandle);
- success = true;
- } catch (KeeperException.NoNodeException e) {
// We wrap the exception here so that it could be caught in DefaultJobGraphStore
- throw new NotExistException("ZooKeeper node " + path + " does not exist.", e);
+ throw ExceptionUtils.findThrowable(e, KeeperException.NoNodeException.class)
+ .map(
+ nnee ->
+ new NotExistException(
+ "ZooKeeper node " + path + " does not exist.", nnee))
+ .orElseThrow(() -> e);
} finally {
- if (success) {
- oldStateHandle.discardState();
- } else {
- newStateHandle.discardState();
+ if (discardOldState) {
+ discardState(oldStateHandle);
}
+
+ if (discardNewState) {
+ discardState(newStateHandle);
+ }
+ }
+ }
+
+ // this method is provided for the sole purpose of easier testing
+ @VisibleForTesting
+ protected void setStateHandle(String path, byte[] serializedStateHandle, int expectedVersion)
+ throws Exception {
+ // Replace state handle in ZooKeeper.
+ client.setData().withVersion(expectedVersion).forPath(path, serializedStateHandle);
+ }
+
+ private static void discardState(@Nullable StateObject stateObject) throws Exception {
Review comment:
`@Nullable` required?
##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesStateHandleStore.java
##########
@@ -470,6 +508,26 @@ public String toString() {
return this.getClass().getSimpleName() + "{configMapName='" + configMapName + "'}";
}
+ private byte[] serializeStateHandle(RetrievableStateHandle<T> retrievableStateHandle)
+ throws Exception {
+ try {
+ // Serialize the state handle. This writes the state to the backend.
+ return InstantiationUtil.serializeObject(retrievableStateHandle);
+ } catch (Exception e) {
+ discardState(retrievableStateHandle);
+ ExceptionUtils.rethrow(e);
+ }
+
+ // will never happen but is added to please the compiler
+ return new byte[0];
+ }
+
+ private static void discardState(@Nullable StateObject stateObject) throws Exception {
Review comment:
Is `@Nullable` required?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-832214416
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17556",
"triggerID" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"triggerType" : "PUSH"
}, {
"hash" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"triggerType" : "PUSH"
}, {
"hash" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17565",
"triggerID" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"triggerType" : "PUSH"
}, {
"hash" : "4b34da74d08c5777b4fe69060acfea6738425844",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17854",
"triggerID" : "4b34da74d08c5777b4fe69060acfea6738425844",
"triggerType" : "PUSH"
}, {
"hash" : "44bb7657519abe9578d2b19ed65f668c0538c828",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44bb7657519abe9578d2b19ed65f668c0538c828",
"triggerType" : "PUSH"
}, {
"hash" : "13993329ba9c544413f04d23a8d0038dabc6b638",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17899",
"triggerID" : "13993329ba9c544413f04d23a8d0038dabc6b638",
"triggerType" : "PUSH"
}, {
"hash" : "c9288127b18f00887b82b710bd42fc1df1487753",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17920",
"triggerID" : "c9288127b18f00887b82b710bd42fc1df1487753",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 707261cd4e5b59e7e73fcbd55d3f76c66aebec48 UNKNOWN
* 44bb7657519abe9578d2b19ed65f668c0538c828 UNKNOWN
* 13993329ba9c544413f04d23a8d0038dabc6b638 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17899)
* c9288127b18f00887b82b710bd42fc1df1487753 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17920)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
XComp commented on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-839642597
Thanks @rkhachatryan for your comments. I addressed all of them either through the code or with a response. Please have a look.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
XComp commented on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-832557309
Build failures for [02f469d](https://dev.azure.com/mapohl/flink/_build/results?buildId=408&view=results) and [14a487](https://dev.azure.com/mapohl/flink/_build/results?buildId=409&view=results) have been addressed in the following tickets:
* [FLINK-22568](https://issues.apache.org/jira/browse/FLINK-22568) RescalingITCase.testSavepointRescalingInPartitionedOperatorStateList
* [FLINK-22571](https://issues.apache.org/jira/browse/FLINK-22571) SavepointITCase.testSavepointForJobWithIteration
* [FLINK-22312](https://issues.apache.org/jira/browse/FLINK-22312) YARNSessionFIFOITCase.checkForProhibitedLogContents
* [FLINK-22567](https://issues.apache.org/jira/browse/FLINK-22567) OrcFsStreamingSinkITCase
* [FLINK-22572](https://issues.apache.org/jira/browse/FLINK-22572) JsonFsStreamSinkITCase.testPart
* [FLINK-22566](https://issues.apache.org/jira/browse/FLINK-22566) Running Kerberized YARN application on Docker test e2e test
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a change in pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
XComp commented on a change in pull request #15832:
URL: https://github.com/apache/flink/pull/15832#discussion_r630021216
##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesStateHandleStore.java
##########
@@ -114,21 +118,28 @@ public KubernetesStateHandleStore(
* @param key Key in ConfigMap
* @param state State to be added
* @throws AlreadyExistException if the name already exists
+ * @throws PossibleInconsistentStateException if the write-to-Kubernetes operation failed. This
+ * indicates that it's not clear whether the new state was successfully written to
+ * Kubernetes or not. No state was discarded. Proper error handling has to be applied on the
+ * caller's side.
* @throws Exception if persisting state or writing state handle failed
*/
@Override
- public RetrievableStateHandle<T> addAndLock(String key, T state) throws Exception {
+ public RetrievableStateHandle<T> addAndLock(String key, T state)
+ throws PossibleInconsistentStateException, Exception {
checkNotNull(key, "Key in ConfigMap.");
checkNotNull(state, "State.");
final RetrievableStateHandle<T> storeHandle = storage.store(state);
- boolean success = false;
+ final byte[] serializedStoreHandle = serializeStateHandle(storeHandle);
+ // initialize flag to serve the failure case
+ boolean discardState = true;
try {
- final byte[] serializedStoreHandle = InstantiationUtil.serializeObject(storeHandle);
- success =
- kubeClient
+ // a successful operation will result in the state not being discarded
+ discardState =
+ !kubeClient
Review comment:
I thought about it once more. The initial motivation to change that came from the fact that we are introducing an error case which is handled as if the method was successful. Using the `success` flag in that case would have been misleading. Alternatively, I could have created a new flag to workaround this double meaning of the same flag. But I'm gonna keep it like that for now...
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] tillrohrmann commented on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-842989916
Thanks for updating the PR @XComp. Merging it now.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-832214416
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17556",
"triggerID" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"triggerType" : "PUSH"
}, {
"hash" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"triggerType" : "PUSH"
}, {
"hash" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17565",
"triggerID" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"triggerType" : "PUSH"
}, {
"hash" : "4b34da74d08c5777b4fe69060acfea6738425844",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17854",
"triggerID" : "4b34da74d08c5777b4fe69060acfea6738425844",
"triggerType" : "PUSH"
}, {
"hash" : "44bb7657519abe9578d2b19ed65f668c0538c828",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44bb7657519abe9578d2b19ed65f668c0538c828",
"triggerType" : "PUSH"
}, {
"hash" : "13993329ba9c544413f04d23a8d0038dabc6b638",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17899",
"triggerID" : "13993329ba9c544413f04d23a8d0038dabc6b638",
"triggerType" : "PUSH"
}, {
"hash" : "c9288127b18f00887b82b710bd42fc1df1487753",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17920",
"triggerID" : "c9288127b18f00887b82b710bd42fc1df1487753",
"triggerType" : "PUSH"
}, {
"hash" : "4c7f59bea05d020743c4aa546e569729e2a7a92a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4c7f59bea05d020743c4aa546e569729e2a7a92a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 707261cd4e5b59e7e73fcbd55d3f76c66aebec48 UNKNOWN
* 44bb7657519abe9578d2b19ed65f668c0538c828 UNKNOWN
* 13993329ba9c544413f04d23a8d0038dabc6b638 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17899)
* c9288127b18f00887b82b710bd42fc1df1487753 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17920)
* 4c7f59bea05d020743c4aa546e569729e2a7a92a UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-832214416
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17556",
"triggerID" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"triggerType" : "PUSH"
}, {
"hash" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a14a487a208ab8c5ea047665e472fd27f829e51c Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17556)
* 707261cd4e5b59e7e73fcbd55d3f76c66aebec48 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a change in pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
XComp commented on a change in pull request #15832:
URL: https://github.com/apache/flink/pull/15832#discussion_r630796139
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperStateHandleStore.java
##########
@@ -126,56 +144,71 @@ public ZooKeeperStateHandleStore(
* @param pathInZooKeeper Destination path in ZooKeeper (expected to *not* exist yet)
* @param state State to be added
* @return The Created {@link RetrievableStateHandle}.
+ * @throws PossibleInconsistentStateException if the write-to-ZooKeeper operation failed. This
+ * indicates that it's not clear whether the new state was successfully written to ZooKeeper
+ * or not. Proper error handling has to be applied on the caller's side.
* @throws Exception If a ZooKeeper or state handle operation fails
*/
@Override
- public RetrievableStateHandle<T> addAndLock(String pathInZooKeeper, T state) throws Exception {
+ public RetrievableStateHandle<T> addAndLock(String pathInZooKeeper, T state)
+ throws PossibleInconsistentStateException, Exception {
checkNotNull(pathInZooKeeper, "Path in ZooKeeper");
checkNotNull(state, "State");
final String path = normalizePath(pathInZooKeeper);
RetrievableStateHandle<T> storeHandle = storage.store(state);
- boolean success = false;
+ byte[] serializedStoreHandle = serializeStateHandle(storeHandle);
+ // initialize flag to serve the failure case
+ boolean discardState = true;
try {
- // Serialize the state handle. This writes the state to the backend.
- byte[] serializedStoreHandle = InstantiationUtil.serializeObject(storeHandle);
-
- // Write state handle (not the actual state) to ZooKeeper. This is expected to be
- // smaller than the state itself. This level of indirection makes sure that data in
- // ZooKeeper is small, because ZooKeeper is designed for data in the KB range, but
- // the state can be larger.
- // Create the lock node in a transaction with the actual state node. That way we can
- // prevent
- // race conditions with a concurrent delete operation.
- client.inTransaction()
- .create()
- .withMode(CreateMode.PERSISTENT)
- .forPath(path, serializedStoreHandle)
- .and()
- .create()
- .withMode(CreateMode.EPHEMERAL)
- .forPath(getLockPath(path))
- .and()
- .commit();
-
- success = true;
+ writeStoreHandleTransactionally(path, serializedStoreHandle);
+ discardState = false;
return storeHandle;
- } catch (KeeperException.NodeExistsException e) {
+ } catch (Exception e) {
+ if (indicatesPossiblyInconsistentState(e)) {
+ discardState = false;
+ throw new PossibleInconsistentStateException(e);
+ }
+
// We wrap the exception here so that it could be caught in DefaultJobGraphStore
- throw new AlreadyExistException("ZooKeeper node " + path + " already exists.", e);
+ throw ExceptionUtils.findThrowable(e, KeeperException.NodeExistsException.class)
Review comment:
I gave it another check. You are right: The `NodeExistsException` is part of the `ZooKeeperStateHandleStore.PRE_COMMIT_EXCEPTIONS`. But [indicatesPossiblyInconsistentState|https://github.com/apache/flink/pull/15832/files#diff-073205da7813621104fd0a45b6f0cbf13064ec5c3701018290e7b2319c7c9c07R278] checks whether the passed exception is not part of the `PRE_COMMIT_EXCEPTIONS`. I feel that's where the confusion is coming from.
[ZooKeeperStateHandleStoreTest.testAddFailureHandlingForNodeExistsException](https://github.com/apache/flink/pull/15832/files#diff-d8d258ac93acbac35b9d7a136d5de0a5a794eec598119fc1174be42ec0e9b36aR174) handles this code path.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-832214416
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17556",
"triggerID" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"triggerType" : "PUSH"
}, {
"hash" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"triggerType" : "PUSH"
}, {
"hash" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17565",
"triggerID" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"triggerType" : "PUSH"
}, {
"hash" : "4b34da74d08c5777b4fe69060acfea6738425844",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17854",
"triggerID" : "4b34da74d08c5777b4fe69060acfea6738425844",
"triggerType" : "PUSH"
}, {
"hash" : "44bb7657519abe9578d2b19ed65f668c0538c828",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44bb7657519abe9578d2b19ed65f668c0538c828",
"triggerType" : "PUSH"
}, {
"hash" : "13993329ba9c544413f04d23a8d0038dabc6b638",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17899",
"triggerID" : "13993329ba9c544413f04d23a8d0038dabc6b638",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 707261cd4e5b59e7e73fcbd55d3f76c66aebec48 UNKNOWN
* 4b34da74d08c5777b4fe69060acfea6738425844 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17854)
* 44bb7657519abe9578d2b19ed65f668c0538c828 UNKNOWN
* 13993329ba9c544413f04d23a8d0038dabc6b638 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17899)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-832214416
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17556",
"triggerID" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a14a487a208ab8c5ea047665e472fd27f829e51c Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17556)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-832214416
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17556",
"triggerID" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"triggerType" : "PUSH"
}, {
"hash" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"triggerType" : "PUSH"
}, {
"hash" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17565",
"triggerID" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"triggerType" : "PUSH"
}, {
"hash" : "4b34da74d08c5777b4fe69060acfea6738425844",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17854",
"triggerID" : "4b34da74d08c5777b4fe69060acfea6738425844",
"triggerType" : "PUSH"
}, {
"hash" : "44bb7657519abe9578d2b19ed65f668c0538c828",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44bb7657519abe9578d2b19ed65f668c0538c828",
"triggerType" : "PUSH"
}, {
"hash" : "13993329ba9c544413f04d23a8d0038dabc6b638",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17899",
"triggerID" : "13993329ba9c544413f04d23a8d0038dabc6b638",
"triggerType" : "PUSH"
}, {
"hash" : "c9288127b18f00887b82b710bd42fc1df1487753",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17920",
"triggerID" : "c9288127b18f00887b82b710bd42fc1df1487753",
"triggerType" : "PUSH"
}, {
"hash" : "4c7f59bea05d020743c4aa546e569729e2a7a92a",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17928",
"triggerID" : "4c7f59bea05d020743c4aa546e569729e2a7a92a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 707261cd4e5b59e7e73fcbd55d3f76c66aebec48 UNKNOWN
* 44bb7657519abe9578d2b19ed65f668c0538c828 UNKNOWN
* 4c7f59bea05d020743c4aa546e569729e2a7a92a Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17928)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a change in pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
XComp commented on a change in pull request #15832:
URL: https://github.com/apache/flink/pull/15832#discussion_r630899932
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/util/StateHandleStoreUtils.java
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.util;
+
+import org.apache.flink.runtime.persistence.StateHandleStore;
+import org.apache.flink.runtime.state.StateObject;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.InstantiationUtil;
+
+/**
+ * {@code StateHandleStoreUtils} collects utility methods that might be usefule for any {@link
+ * StateHandleStore} implementation.
+ */
+public class StateHandleStoreUtils {
+
+ /**
+ * Serializes the passed {@link StateObject} and discards the state in case of failure.
+ *
+ * @param stateObject the {@code StateObject} that shall be serialized.
+ * @return The serialized version of the passed {@code StateObject}.
+ * @throws Exception if an error occurred during the serialization. The corresponding {@code
+ * StateObject} will be discarded in that case.
+ */
+ public static byte[] serializeStateHandle(StateObject stateObject) throws Exception {
Review comment:
I thought about the same thing but hesitated since it was only a one-liner. I went ahead with your proposal and even added the missing `StateObjectUtilsTest`.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-832214416
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17556",
"triggerID" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"triggerType" : "PUSH"
}, {
"hash" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"triggerType" : "PUSH"
}, {
"hash" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17565",
"triggerID" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"triggerType" : "PUSH"
}, {
"hash" : "4b34da74d08c5777b4fe69060acfea6738425844",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17854",
"triggerID" : "4b34da74d08c5777b4fe69060acfea6738425844",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 707261cd4e5b59e7e73fcbd55d3f76c66aebec48 UNKNOWN
* 4b34da74d08c5777b4fe69060acfea6738425844 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17854)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-832214416
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17556",
"triggerID" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"triggerType" : "PUSH"
}, {
"hash" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"triggerType" : "PUSH"
}, {
"hash" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17565",
"triggerID" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"triggerType" : "PUSH"
}, {
"hash" : "4b34da74d08c5777b4fe69060acfea6738425844",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17854",
"triggerID" : "4b34da74d08c5777b4fe69060acfea6738425844",
"triggerType" : "PUSH"
}, {
"hash" : "44bb7657519abe9578d2b19ed65f668c0538c828",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44bb7657519abe9578d2b19ed65f668c0538c828",
"triggerType" : "PUSH"
}, {
"hash" : "13993329ba9c544413f04d23a8d0038dabc6b638",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "13993329ba9c544413f04d23a8d0038dabc6b638",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 707261cd4e5b59e7e73fcbd55d3f76c66aebec48 UNKNOWN
* 4b34da74d08c5777b4fe69060acfea6738425844 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17854)
* 44bb7657519abe9578d2b19ed65f668c0538c828 UNKNOWN
* 13993329ba9c544413f04d23a8d0038dabc6b638 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a change in pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
XComp commented on a change in pull request #15832:
URL: https://github.com/apache/flink/pull/15832#discussion_r630810712
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperStateHandleStore.java
##########
@@ -196,29 +229,55 @@ public void replace(String pathInZooKeeper, IntegerResourceVersion expectedVersi
RetrievableStateHandle<T> newStateHandle = storage.store(state);
- boolean success = false;
+ final byte[] serializedStateHandle = serializeStateHandle(newStateHandle);
+ // initialize flags to serve the failure case
+ boolean discardOldState = false;
+ boolean discardNewState = true;
try {
- // Serialize the new state handle. This writes the state to the backend.
- byte[] serializedStateHandle = InstantiationUtil.serializeObject(newStateHandle);
+ setStateHandle(path, serializedStateHandle, expectedVersion.getValue());
+
+ // swap subject for deletion in case of success
+ discardOldState = true;
+ discardNewState = false;
+ } catch (Exception e) {
+ if (indicatesPossiblyInconsistentState(e)) {
+ // it's unclear whether the state handle metadata was written to ZooKeeper -
+ // hence, we don't discard any data
+ discardNewState = false;
+ throw new PossibleInconsistentStateException(e);
+ }
- // Replace state handle in ZooKeeper.
- client.setData()
- .withVersion(expectedVersion.getValue())
- .forPath(path, serializedStateHandle);
- success = true;
- } catch (KeeperException.NoNodeException e) {
// We wrap the exception here so that it could be caught in DefaultJobGraphStore
- throw new NotExistException("ZooKeeper node " + path + " does not exist.", e);
+ throw ExceptionUtils.findThrowable(e, KeeperException.NoNodeException.class)
+ .map(
+ nnee ->
+ new NotExistException(
+ "ZooKeeper node " + path + " does not exist.", nnee))
+ .orElseThrow(() -> e);
} finally {
- if (success) {
+ if (discardOldState) {
oldStateHandle.discardState();
- } else {
+ }
+
+ if (discardNewState) {
newStateHandle.discardState();
}
}
}
+ // this method is provided for the sole purpose of easier testing
+ @VisibleForTesting
+ protected void setStateHandle(String path, byte[] serializedStateHandle, int expectedVersion)
+ throws Exception {
+ // Replace state handle in ZooKeeper.
+ client.setData().withVersion(expectedVersion).forPath(path, serializedStateHandle);
+ }
+
+ private boolean indicatesPossiblyInconsistentState(Exception e) {
+ return !PRE_COMMIT_EXCEPTIONS.contains(e.getClass());
Review comment:
I would like to keep it more explicit. If subclasses are meant to be in that set, we should add them explicitly. Your example with `AuthExpiredException` is just a made up example, isn't it? I couldn't find it in the code.
For this change, I went through the error cases that are listed in [KeeperException.create](https://github.com/apache/zookeeper/blob/branch-3.4/zookeeper-server/src/main/java/org/apache/zookeeper/KeeperException.java#L93) to cover the possible cases that are triggered by the `commit()` call.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-832214416
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17556",
"triggerID" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"triggerType" : "PUSH"
}, {
"hash" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"triggerType" : "PUSH"
}, {
"hash" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a14a487a208ab8c5ea047665e472fd27f829e51c Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17556)
* 707261cd4e5b59e7e73fcbd55d3f76c66aebec48 UNKNOWN
* 7fad07e7c67e05691fdf158173e84017c05cbeb0 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-832214416
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17556",
"triggerID" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"triggerType" : "PUSH"
}, {
"hash" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"triggerType" : "PUSH"
}, {
"hash" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17565",
"triggerID" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"triggerType" : "PUSH"
}, {
"hash" : "4b34da74d08c5777b4fe69060acfea6738425844",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17854",
"triggerID" : "4b34da74d08c5777b4fe69060acfea6738425844",
"triggerType" : "PUSH"
}, {
"hash" : "44bb7657519abe9578d2b19ed65f668c0538c828",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44bb7657519abe9578d2b19ed65f668c0538c828",
"triggerType" : "PUSH"
}, {
"hash" : "13993329ba9c544413f04d23a8d0038dabc6b638",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17899",
"triggerID" : "13993329ba9c544413f04d23a8d0038dabc6b638",
"triggerType" : "PUSH"
}, {
"hash" : "c9288127b18f00887b82b710bd42fc1df1487753",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "c9288127b18f00887b82b710bd42fc1df1487753",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 707261cd4e5b59e7e73fcbd55d3f76c66aebec48 UNKNOWN
* 44bb7657519abe9578d2b19ed65f668c0538c828 UNKNOWN
* 13993329ba9c544413f04d23a8d0038dabc6b638 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17899)
* c9288127b18f00887b82b710bd42fc1df1487753 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
XComp commented on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-838190014
Thanks @tillrohrmann for reviewing the changes. I applied your comments and rebased the branch.
@rkhachatryan Could you give it a pass as well since I touched some code related to Checkpoints (namely refactoring the cleaner a bit). The commits that might be relevant to you are [01969be](https://github.com/apache/flink/pull/15832/commits/01969becd2f152c5d70f07d61376e630d52c0a23), [3b4a81f](https://github.com/apache/flink/pull/15832/commits/3b4a81fa658c4a094ed5654a3b802379144be532) and [4b34da7](https://github.com/apache/flink/pull/15832/commits/4b34da74d08c5777b4fe69060acfea6738425844)
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a change in pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
XComp commented on a change in pull request #15832:
URL: https://github.com/apache/flink/pull/15832#discussion_r630796139
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperStateHandleStore.java
##########
@@ -126,56 +144,71 @@ public ZooKeeperStateHandleStore(
* @param pathInZooKeeper Destination path in ZooKeeper (expected to *not* exist yet)
* @param state State to be added
* @return The Created {@link RetrievableStateHandle}.
+ * @throws PossibleInconsistentStateException if the write-to-ZooKeeper operation failed. This
+ * indicates that it's not clear whether the new state was successfully written to ZooKeeper
+ * or not. Proper error handling has to be applied on the caller's side.
* @throws Exception If a ZooKeeper or state handle operation fails
*/
@Override
- public RetrievableStateHandle<T> addAndLock(String pathInZooKeeper, T state) throws Exception {
+ public RetrievableStateHandle<T> addAndLock(String pathInZooKeeper, T state)
+ throws PossibleInconsistentStateException, Exception {
checkNotNull(pathInZooKeeper, "Path in ZooKeeper");
checkNotNull(state, "State");
final String path = normalizePath(pathInZooKeeper);
RetrievableStateHandle<T> storeHandle = storage.store(state);
- boolean success = false;
+ byte[] serializedStoreHandle = serializeStateHandle(storeHandle);
+ // initialize flag to serve the failure case
+ boolean discardState = true;
try {
- // Serialize the state handle. This writes the state to the backend.
- byte[] serializedStoreHandle = InstantiationUtil.serializeObject(storeHandle);
-
- // Write state handle (not the actual state) to ZooKeeper. This is expected to be
- // smaller than the state itself. This level of indirection makes sure that data in
- // ZooKeeper is small, because ZooKeeper is designed for data in the KB range, but
- // the state can be larger.
- // Create the lock node in a transaction with the actual state node. That way we can
- // prevent
- // race conditions with a concurrent delete operation.
- client.inTransaction()
- .create()
- .withMode(CreateMode.PERSISTENT)
- .forPath(path, serializedStoreHandle)
- .and()
- .create()
- .withMode(CreateMode.EPHEMERAL)
- .forPath(getLockPath(path))
- .and()
- .commit();
-
- success = true;
+ writeStoreHandleTransactionally(path, serializedStoreHandle);
+ discardState = false;
return storeHandle;
- } catch (KeeperException.NodeExistsException e) {
+ } catch (Exception e) {
+ if (indicatesPossiblyInconsistentState(e)) {
+ discardState = false;
+ throw new PossibleInconsistentStateException(e);
+ }
+
// We wrap the exception here so that it could be caught in DefaultJobGraphStore
- throw new AlreadyExistException("ZooKeeper node " + path + " already exists.", e);
+ throw ExceptionUtils.findThrowable(e, KeeperException.NodeExistsException.class)
Review comment:
I gave it another check. You are right: The `NodeExistsException` is part of the `ZooKeeperStateHandleStore.PRE_COMMIT_EXCEPTIONS`. But [indicatesPossiblyInconsistentState|https://github.com/apache/flink/pull/15832/files#diff-073205da7813621104fd0a45b6f0cbf13064ec5c3701018290e7b2319c7c9c07R278] checks whether the passed exception is not part of the `PRE_COMMIT_EXCEPTIONS`. I feel that's where the confusion is coming from.
[ZooKeeperStateHandleStoreTest.testAddFailureHandlingForNodeExistsException|https://github.com/apache/flink/pull/15832/files#diff-d8d258ac93acbac35b9d7a136d5de0a5a794eec598119fc1174be42ec0e9b36aR174] handles this code path.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-832197189
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 4c7f59bea05d020743c4aa546e569729e2a7a92a (Sat Aug 28 12:09:35 UTC 2021)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<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] rkhachatryan commented on a change in pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #15832:
URL: https://github.com/apache/flink/pull/15832#discussion_r630873085
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperStateHandleStore.java
##########
@@ -196,29 +229,55 @@ public void replace(String pathInZooKeeper, IntegerResourceVersion expectedVersi
RetrievableStateHandle<T> newStateHandle = storage.store(state);
- boolean success = false;
+ final byte[] serializedStateHandle = serializeStateHandle(newStateHandle);
+ // initialize flags to serve the failure case
+ boolean discardOldState = false;
+ boolean discardNewState = true;
try {
- // Serialize the new state handle. This writes the state to the backend.
- byte[] serializedStateHandle = InstantiationUtil.serializeObject(newStateHandle);
+ setStateHandle(path, serializedStateHandle, expectedVersion.getValue());
+
+ // swap subject for deletion in case of success
+ discardOldState = true;
+ discardNewState = false;
+ } catch (Exception e) {
+ if (indicatesPossiblyInconsistentState(e)) {
+ // it's unclear whether the state handle metadata was written to ZooKeeper -
+ // hence, we don't discard any data
+ discardNewState = false;
+ throw new PossibleInconsistentStateException(e);
+ }
- // Replace state handle in ZooKeeper.
- client.setData()
- .withVersion(expectedVersion.getValue())
- .forPath(path, serializedStateHandle);
- success = true;
- } catch (KeeperException.NoNodeException e) {
// We wrap the exception here so that it could be caught in DefaultJobGraphStore
- throw new NotExistException("ZooKeeper node " + path + " does not exist.", e);
+ throw ExceptionUtils.findThrowable(e, KeeperException.NoNodeException.class)
+ .map(
+ nnee ->
+ new NotExistException(
+ "ZooKeeper node " + path + " does not exist.", nnee))
+ .orElseThrow(() -> e);
} finally {
- if (success) {
+ if (discardOldState) {
oldStateHandle.discardState();
- } else {
+ }
+
+ if (discardNewState) {
newStateHandle.discardState();
}
}
}
+ // this method is provided for the sole purpose of easier testing
+ @VisibleForTesting
+ protected void setStateHandle(String path, byte[] serializedStateHandle, int expectedVersion)
+ throws Exception {
+ // Replace state handle in ZooKeeper.
+ client.setData().withVersion(expectedVersion).forPath(path, serializedStateHandle);
+ }
+
+ private boolean indicatesPossiblyInconsistentState(Exception e) {
+ return !PRE_COMMIT_EXCEPTIONS.contains(e.getClass());
Review comment:
Fair enough (yes, `AuthExpiredException` is just a made up example).
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-832214416
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17556",
"triggerID" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"triggerType" : "PUSH"
}, {
"hash" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"triggerType" : "PUSH"
}, {
"hash" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17565",
"triggerID" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"triggerType" : "PUSH"
}, {
"hash" : "4b34da74d08c5777b4fe69060acfea6738425844",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17854",
"triggerID" : "4b34da74d08c5777b4fe69060acfea6738425844",
"triggerType" : "PUSH"
}, {
"hash" : "44bb7657519abe9578d2b19ed65f668c0538c828",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44bb7657519abe9578d2b19ed65f668c0538c828",
"triggerType" : "PUSH"
}, {
"hash" : "13993329ba9c544413f04d23a8d0038dabc6b638",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17899",
"triggerID" : "13993329ba9c544413f04d23a8d0038dabc6b638",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 707261cd4e5b59e7e73fcbd55d3f76c66aebec48 UNKNOWN
* 44bb7657519abe9578d2b19ed65f668c0538c828 UNKNOWN
* 13993329ba9c544413f04d23a8d0038dabc6b638 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17899)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a change in pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
XComp commented on a change in pull request #15832:
URL: https://github.com/apache/flink/pull/15832#discussion_r629976027
##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesStateHandleStore.java
##########
@@ -470,6 +508,26 @@ public String toString() {
return this.getClass().getSimpleName() + "{configMapName='" + configMapName + "'}";
}
+ private byte[] serializeStateHandle(RetrievableStateHandle<T> retrievableStateHandle)
+ throws Exception {
+ try {
+ // Serialize the state handle. This writes the state to the backend.
+ return InstantiationUtil.serializeObject(retrievableStateHandle);
+ } catch (Exception e) {
+ discardState(retrievableStateHandle);
+ ExceptionUtils.rethrow(e);
+ }
+
+ // will never happen but is added to please the compiler
+ return new byte[0];
+ }
+
+ private static void discardState(@Nullable StateObject stateObject) throws Exception {
Review comment:
I was guided by one null check [here](https://github.com/apache/flink/pull/15832/files#diff-ddcc7a8f5b1139881124c01fa4f0c44a1e49c29b611a9cd71103790ffd66d7c3L159). But I double-checked now: The `null` check is not necessary. I removed it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a change in pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
XComp commented on a change in pull request #15832:
URL: https://github.com/apache/flink/pull/15832#discussion_r626294245
##########
File path: flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClientTest.java
##########
@@ -504,6 +550,7 @@ private KubernetesConfigMap buildTestingConfigMap() {
.withNewMetadata()
.withName(TESTING_CONFIG_MAP_NAME)
.withLabels(TESTING_LABELS)
+ .withNamespace(NAMESPACE)
Review comment:
I added this change as it makes it more handy when having the namespace also specified in the ConfigMap. See [KubernetesClientTestBase.mockConfigMapRequest](https://github.com/XComp/flink/blob/2e475a777ebebbbcee47eaafaa14c3ccf2724367/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClientTestBase.java#L77). It's not really necessary since `NAMESPACE` is also available in that class. But I ran into the issue of the missing namespace first before discovering that the namespace is set through a static variable.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a change in pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
XComp commented on a change in pull request #15832:
URL: https://github.com/apache/flink/pull/15832#discussion_r626296284
##########
File path: flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesStateHandleStoreITCase.java
##########
@@ -68,8 +68,8 @@ public void testMultipleKubernetesStateHandleStores() throws Exception {
new TestingLeaderCallbackHandler[leaderNum];
@SuppressWarnings("unchecked")
- final KubernetesStateHandleStore<Long>[] stateHandleStores =
- new KubernetesStateHandleStore[leaderNum];
+ final KubernetesStateHandleStore<TestingLongStateHandleHelper.LongStateHandle>[]
Review comment:
The Kubernetes-related ITCases are currently not running. I created [FLINK-20564](https://issues.apache.org/jira/browse/FLINK-22564) to cover this. ~I haven't had the chance to run the ITCase locally, yet.~ I checked that the test passes locally including my changes.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a change in pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
XComp commented on a change in pull request #15832:
URL: https://github.com/apache/flink/pull/15832#discussion_r630796139
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperStateHandleStore.java
##########
@@ -126,56 +144,71 @@ public ZooKeeperStateHandleStore(
* @param pathInZooKeeper Destination path in ZooKeeper (expected to *not* exist yet)
* @param state State to be added
* @return The Created {@link RetrievableStateHandle}.
+ * @throws PossibleInconsistentStateException if the write-to-ZooKeeper operation failed. This
+ * indicates that it's not clear whether the new state was successfully written to ZooKeeper
+ * or not. Proper error handling has to be applied on the caller's side.
* @throws Exception If a ZooKeeper or state handle operation fails
*/
@Override
- public RetrievableStateHandle<T> addAndLock(String pathInZooKeeper, T state) throws Exception {
+ public RetrievableStateHandle<T> addAndLock(String pathInZooKeeper, T state)
+ throws PossibleInconsistentStateException, Exception {
checkNotNull(pathInZooKeeper, "Path in ZooKeeper");
checkNotNull(state, "State");
final String path = normalizePath(pathInZooKeeper);
RetrievableStateHandle<T> storeHandle = storage.store(state);
- boolean success = false;
+ byte[] serializedStoreHandle = serializeStateHandle(storeHandle);
+ // initialize flag to serve the failure case
+ boolean discardState = true;
try {
- // Serialize the state handle. This writes the state to the backend.
- byte[] serializedStoreHandle = InstantiationUtil.serializeObject(storeHandle);
-
- // Write state handle (not the actual state) to ZooKeeper. This is expected to be
- // smaller than the state itself. This level of indirection makes sure that data in
- // ZooKeeper is small, because ZooKeeper is designed for data in the KB range, but
- // the state can be larger.
- // Create the lock node in a transaction with the actual state node. That way we can
- // prevent
- // race conditions with a concurrent delete operation.
- client.inTransaction()
- .create()
- .withMode(CreateMode.PERSISTENT)
- .forPath(path, serializedStoreHandle)
- .and()
- .create()
- .withMode(CreateMode.EPHEMERAL)
- .forPath(getLockPath(path))
- .and()
- .commit();
-
- success = true;
+ writeStoreHandleTransactionally(path, serializedStoreHandle);
+ discardState = false;
return storeHandle;
- } catch (KeeperException.NodeExistsException e) {
+ } catch (Exception e) {
+ if (indicatesPossiblyInconsistentState(e)) {
+ discardState = false;
+ throw new PossibleInconsistentStateException(e);
+ }
+
// We wrap the exception here so that it could be caught in DefaultJobGraphStore
- throw new AlreadyExistException("ZooKeeper node " + path + " already exists.", e);
+ throw ExceptionUtils.findThrowable(e, KeeperException.NodeExistsException.class)
Review comment:
I gave it another check. You are right: The `NodeExistsException` is part of the `ZooKeeperStateHandleStore.PRE_COMMIT_EXCEPTIONS`. But [indicatesPossiblyInconsistentState](https://github.com/apache/flink/pull/15832/files#diff-073205da7813621104fd0a45b6f0cbf13064ec5c3701018290e7b2319c7c9c07R278) checks whether the passed exception is not part of the `PRE_COMMIT_EXCEPTIONS`. I feel that's where the confusion is coming from.
[ZooKeeperStateHandleStoreTest.testAddFailureHandlingForNodeExistsException](https://github.com/apache/flink/pull/15832/files#diff-d8d258ac93acbac35b9d7a136d5de0a5a794eec598119fc1174be42ec0e9b36aR174) handles this code path.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
XComp commented on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-841813005
The [reported build failure](https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=17928&view=logs&j=b2f046ab-ae17-5406-acdc-240be7e870e4&t=93e5ae06-d194-513d-ba8d-150ef6da1d7c&l=9225) happened due to the unrelated issue [FLINK-22595](https://issues.apache.org/jira/browse/FLINK-22595?focusedCommentId=17345694&page=com.atlassian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel#comment-17345694)
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-832214416
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17556",
"triggerID" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"triggerType" : "PUSH"
}, {
"hash" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"triggerType" : "PUSH"
}, {
"hash" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17565",
"triggerID" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"triggerType" : "PUSH"
}, {
"hash" : "4b34da74d08c5777b4fe69060acfea6738425844",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17854",
"triggerID" : "4b34da74d08c5777b4fe69060acfea6738425844",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 707261cd4e5b59e7e73fcbd55d3f76c66aebec48 UNKNOWN
* 7fad07e7c67e05691fdf158173e84017c05cbeb0 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17565)
* 4b34da74d08c5777b4fe69060acfea6738425844 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17854)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a change in pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
XComp commented on a change in pull request #15832:
URL: https://github.com/apache/flink/pull/15832#discussion_r630799380
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperStateHandleStore.java
##########
@@ -126,56 +144,71 @@ public ZooKeeperStateHandleStore(
* @param pathInZooKeeper Destination path in ZooKeeper (expected to *not* exist yet)
* @param state State to be added
* @return The Created {@link RetrievableStateHandle}.
+ * @throws PossibleInconsistentStateException if the write-to-ZooKeeper operation failed. This
+ * indicates that it's not clear whether the new state was successfully written to ZooKeeper
+ * or not. Proper error handling has to be applied on the caller's side.
* @throws Exception If a ZooKeeper or state handle operation fails
*/
@Override
- public RetrievableStateHandle<T> addAndLock(String pathInZooKeeper, T state) throws Exception {
+ public RetrievableStateHandle<T> addAndLock(String pathInZooKeeper, T state)
+ throws PossibleInconsistentStateException, Exception {
checkNotNull(pathInZooKeeper, "Path in ZooKeeper");
checkNotNull(state, "State");
final String path = normalizePath(pathInZooKeeper);
RetrievableStateHandle<T> storeHandle = storage.store(state);
- boolean success = false;
+ byte[] serializedStoreHandle = serializeStateHandle(storeHandle);
+ // initialize flag to serve the failure case
+ boolean discardState = true;
try {
- // Serialize the state handle. This writes the state to the backend.
- byte[] serializedStoreHandle = InstantiationUtil.serializeObject(storeHandle);
-
- // Write state handle (not the actual state) to ZooKeeper. This is expected to be
- // smaller than the state itself. This level of indirection makes sure that data in
- // ZooKeeper is small, because ZooKeeper is designed for data in the KB range, but
- // the state can be larger.
- // Create the lock node in a transaction with the actual state node. That way we can
- // prevent
- // race conditions with a concurrent delete operation.
- client.inTransaction()
- .create()
- .withMode(CreateMode.PERSISTENT)
- .forPath(path, serializedStoreHandle)
- .and()
- .create()
- .withMode(CreateMode.EPHEMERAL)
- .forPath(getLockPath(path))
- .and()
- .commit();
-
- success = true;
+ writeStoreHandleTransactionally(path, serializedStoreHandle);
+ discardState = false;
return storeHandle;
- } catch (KeeperException.NodeExistsException e) {
+ } catch (Exception e) {
+ if (indicatesPossiblyInconsistentState(e)) {
+ discardState = false;
+ throw new PossibleInconsistentStateException(e);
+ }
+
// We wrap the exception here so that it could be caught in DefaultJobGraphStore
- throw new AlreadyExistException("ZooKeeper node " + path + " already exists.", e);
+ throw ExceptionUtils.findThrowable(e, KeeperException.NodeExistsException.class)
+ .map(
+ nee ->
+ new AlreadyExistException(
+ "ZooKeeper node " + path + " already exists.", nee))
+ .orElseThrow(() -> e);
} finally {
- if (!success) {
- // Cleanup the state handle if it was not written to ZooKeeper.
- if (storeHandle != null) {
- storeHandle.discardState();
- }
+ if (discardState) {
+ storeHandle.discardState();
Review comment:
That's a good catch. Thanks for the idea. I will apply it. 👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] tillrohrmann closed pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
tillrohrmann closed pull request #15832:
URL: https://github.com/apache/flink/pull/15832
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-832214416
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17556",
"triggerID" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"triggerType" : "PUSH"
}, {
"hash" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"triggerType" : "PUSH"
}, {
"hash" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17565",
"triggerID" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"triggerType" : "PUSH"
}, {
"hash" : "4b34da74d08c5777b4fe69060acfea6738425844",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17854",
"triggerID" : "4b34da74d08c5777b4fe69060acfea6738425844",
"triggerType" : "PUSH"
}, {
"hash" : "44bb7657519abe9578d2b19ed65f668c0538c828",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44bb7657519abe9578d2b19ed65f668c0538c828",
"triggerType" : "PUSH"
}, {
"hash" : "13993329ba9c544413f04d23a8d0038dabc6b638",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17899",
"triggerID" : "13993329ba9c544413f04d23a8d0038dabc6b638",
"triggerType" : "PUSH"
}, {
"hash" : "c9288127b18f00887b82b710bd42fc1df1487753",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17920",
"triggerID" : "c9288127b18f00887b82b710bd42fc1df1487753",
"triggerType" : "PUSH"
}, {
"hash" : "4c7f59bea05d020743c4aa546e569729e2a7a92a",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17928",
"triggerID" : "4c7f59bea05d020743c4aa546e569729e2a7a92a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 707261cd4e5b59e7e73fcbd55d3f76c66aebec48 UNKNOWN
* 44bb7657519abe9578d2b19ed65f668c0538c828 UNKNOWN
* 13993329ba9c544413f04d23a8d0038dabc6b638 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17899)
* c9288127b18f00887b82b710bd42fc1df1487753 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17920)
* 4c7f59bea05d020743c4aa546e569729e2a7a92a Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17928)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-832214416
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17556",
"triggerID" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"triggerType" : "PUSH"
}, {
"hash" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"triggerType" : "PUSH"
}, {
"hash" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17565",
"triggerID" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a14a487a208ab8c5ea047665e472fd27f829e51c Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17556)
* 707261cd4e5b59e7e73fcbd55d3f76c66aebec48 UNKNOWN
* 7fad07e7c67e05691fdf158173e84017c05cbeb0 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17565)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] tillrohrmann commented on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-836866043
Feel free to also open the PRs for the backports so that we get CI.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-832214416
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17556",
"triggerID" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"triggerType" : "PUSH"
}, {
"hash" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"triggerType" : "PUSH"
}, {
"hash" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17565",
"triggerID" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"triggerType" : "PUSH"
}, {
"hash" : "4b34da74d08c5777b4fe69060acfea6738425844",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17854",
"triggerID" : "4b34da74d08c5777b4fe69060acfea6738425844",
"triggerType" : "PUSH"
}, {
"hash" : "44bb7657519abe9578d2b19ed65f668c0538c828",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44bb7657519abe9578d2b19ed65f668c0538c828",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 707261cd4e5b59e7e73fcbd55d3f76c66aebec48 UNKNOWN
* 4b34da74d08c5777b4fe69060acfea6738425844 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17854)
* 44bb7657519abe9578d2b19ed65f668c0538c828 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
XComp commented on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-841813005
The [reported build failure](https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=17928&view=logs&j=b2f046ab-ae17-5406-acdc-240be7e870e4&t=93e5ae06-d194-513d-ba8d-150ef6da1d7c&l=9225) happened due to the unrelated issue [FLINK-22595](https://issues.apache.org/jira/browse/FLINK-22595?focusedCommentId=17345694&page=com.atlassian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel#comment-17345694)
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a change in pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
XComp commented on a change in pull request #15832:
URL: https://github.com/apache/flink/pull/15832#discussion_r630810712
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperStateHandleStore.java
##########
@@ -196,29 +229,55 @@ public void replace(String pathInZooKeeper, IntegerResourceVersion expectedVersi
RetrievableStateHandle<T> newStateHandle = storage.store(state);
- boolean success = false;
+ final byte[] serializedStateHandle = serializeStateHandle(newStateHandle);
+ // initialize flags to serve the failure case
+ boolean discardOldState = false;
+ boolean discardNewState = true;
try {
- // Serialize the new state handle. This writes the state to the backend.
- byte[] serializedStateHandle = InstantiationUtil.serializeObject(newStateHandle);
+ setStateHandle(path, serializedStateHandle, expectedVersion.getValue());
+
+ // swap subject for deletion in case of success
+ discardOldState = true;
+ discardNewState = false;
+ } catch (Exception e) {
+ if (indicatesPossiblyInconsistentState(e)) {
+ // it's unclear whether the state handle metadata was written to ZooKeeper -
+ // hence, we don't discard any data
+ discardNewState = false;
+ throw new PossibleInconsistentStateException(e);
+ }
- // Replace state handle in ZooKeeper.
- client.setData()
- .withVersion(expectedVersion.getValue())
- .forPath(path, serializedStateHandle);
- success = true;
- } catch (KeeperException.NoNodeException e) {
// We wrap the exception here so that it could be caught in DefaultJobGraphStore
- throw new NotExistException("ZooKeeper node " + path + " does not exist.", e);
+ throw ExceptionUtils.findThrowable(e, KeeperException.NoNodeException.class)
+ .map(
+ nnee ->
+ new NotExistException(
+ "ZooKeeper node " + path + " does not exist.", nnee))
+ .orElseThrow(() -> e);
} finally {
- if (success) {
+ if (discardOldState) {
oldStateHandle.discardState();
- } else {
+ }
+
+ if (discardNewState) {
newStateHandle.discardState();
}
}
}
+ // this method is provided for the sole purpose of easier testing
+ @VisibleForTesting
+ protected void setStateHandle(String path, byte[] serializedStateHandle, int expectedVersion)
+ throws Exception {
+ // Replace state handle in ZooKeeper.
+ client.setData().withVersion(expectedVersion).forPath(path, serializedStateHandle);
+ }
+
+ private boolean indicatesPossiblyInconsistentState(Exception e) {
+ return !PRE_COMMIT_EXCEPTIONS.contains(e.getClass());
Review comment:
I would like to keep it more explicit. If subclasses are meant to be in that set, we should add them explicitly. Your example with `AuthExpiredException` is just a made up example, isn't it? I couldn't find it in the code. I was going through the error cases that are listed in [KeeperException.create|https://github.com/apache/zookeeper/blob/branch-3.4/zookeeper-server/src/main/java/org/apache/zookeeper/KeeperException.java#L93] to cover the possible cases that are triggered by the `commit()` call.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a change in pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
XComp commented on a change in pull request #15832:
URL: https://github.com/apache/flink/pull/15832#discussion_r630818299
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorFailureTest.java
##########
@@ -160,8 +177,94 @@ public void testFailingCompletedCheckpointStoreAdd() throws Exception {
.discardState();
}
+ @Test
+ public void testCleanupForGenericFailure() throws Exception {
+ testStoringFailureHandling(new FlinkRuntimeException("Expected exception"), 1);
+ }
+
+ @Test
+ public void testCleanupOmissionForPossibleInconsistentStateException() throws Exception {
+ testStoringFailureHandling(new PossibleInconsistentStateException(), 0);
+ }
+
+ private void testStoringFailureHandling(Exception failure, int expectedCleanupCalls)
+ throws Exception {
+ final JobVertexID jobVertexID1 = new JobVertexID();
+
+ final ExecutionGraph graph =
+ new CheckpointCoordinatorTestingUtils.CheckpointExecutionGraphBuilder()
+ .addJobVertex(jobVertexID1)
+ .build();
+
+ final ExecutionVertex vertex = graph.getJobVertex(jobVertexID1).getTaskVertices()[0];
+ final ExecutionAttemptID attemptId = vertex.getCurrentExecutionAttempt().getAttemptId();
+
+ final StandaloneCheckpointIDCounter checkpointIDCounter =
+ new StandaloneCheckpointIDCounter();
+
+ final ManuallyTriggeredScheduledExecutor manuallyTriggeredScheduledExecutor =
+ new ManuallyTriggeredScheduledExecutor();
+
+ final CompletedCheckpointStore completedCheckpointStore =
+ new FailingCompletedCheckpointStore(
+ (checkpoint, ignoredCleaner, ignoredPostCleanCallback) -> {
+ throw failure;
+ });
+
+ final AtomicInteger cleanupCallCount = new AtomicInteger(0);
+ final CheckpointCoordinator checkpointCoordinator =
+ new CheckpointCoordinatorBuilder()
+ .setExecutionGraph(graph)
+ .setCheckpointIDCounter(checkpointIDCounter)
+ .setCheckpointsCleaner(
+ new CheckpointsCleaner() {
+
+ private static final long serialVersionUID =
+ 2029876992397573325L;
+
+ @Override
+ public void cleanCheckpointOnFailedStoring(
+ CompletedCheckpoint completedCheckpoint,
+ Executor executor) {
+ cleanupCallCount.incrementAndGet();
+ super.cleanCheckpointOnFailedStoring(
+ completedCheckpoint, executor);
+ }
+ })
+ .setCompletedCheckpointStore(completedCheckpointStore)
+ .setTimer(manuallyTriggeredScheduledExecutor)
+ .build();
+ checkpointCoordinator.triggerSavepoint(tmpFolder.newFolder().getAbsolutePath());
+ manuallyTriggeredScheduledExecutor.triggerAll();
+
+ try {
+ checkpointCoordinator.receiveAcknowledgeMessage(
+ new AcknowledgeCheckpoint(
+ graph.getJobID(), attemptId, checkpointIDCounter.getLast()),
+ "unknown location");
+ fail("CheckpointException should have been thrown.");
+ } catch (CheckpointException e) {
+ assertThat(
+ e.getCheckpointFailureReason(),
+ is(CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE));
+ }
+
+ assertThat(cleanupCallCount.get(), is(expectedCleanupCalls));
+ }
+
private static final class FailingCompletedCheckpointStore implements CompletedCheckpointStore {
+ private final TriConsumerWithException<
+ CompletedCheckpoint, CheckpointsCleaner, Runnable, Exception>
+ addCheckpointConsumer;
+
+ public FailingCompletedCheckpointStore(
+ TriConsumerWithException<
+ CompletedCheckpoint, CheckpointsCleaner, Runnable, Exception>
+ addCheckpointConsumer) {
+ this.addCheckpointConsumer = addCheckpointConsumer;
Review comment:
You're right. It would be sufficient enough considering that `FailingCompletedCheckpointStore` is only used internally and does not need this extended flexibility. It was just a habit to generalize these kind of builders. But there is no need to do that considering that it's not used anywhere else. I'm gonna revert that change and go with your proposal.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-832214416
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a14a487a208ab8c5ea047665e472fd27f829e51c UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-832197189
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 a14a487a208ab8c5ea047665e472fd27f829e51c (Tue May 04 19:44:16 UTC 2021)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15832:
URL: https://github.com/apache/flink/pull/15832#issuecomment-832214416
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17556",
"triggerID" : "a14a487a208ab8c5ea047665e472fd27f829e51c",
"triggerType" : "PUSH"
}, {
"hash" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "707261cd4e5b59e7e73fcbd55d3f76c66aebec48",
"triggerType" : "PUSH"
}, {
"hash" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17565",
"triggerID" : "7fad07e7c67e05691fdf158173e84017c05cbeb0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 707261cd4e5b59e7e73fcbd55d3f76c66aebec48 UNKNOWN
* 7fad07e7c67e05691fdf158173e84017c05cbeb0 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=17565)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a change in pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
XComp commented on a change in pull request #15832:
URL: https://github.com/apache/flink/pull/15832#discussion_r626296284
##########
File path: flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesStateHandleStoreITCase.java
##########
@@ -68,8 +68,8 @@ public void testMultipleKubernetesStateHandleStores() throws Exception {
new TestingLeaderCallbackHandler[leaderNum];
@SuppressWarnings("unchecked")
- final KubernetesStateHandleStore<Long>[] stateHandleStores =
- new KubernetesStateHandleStore[leaderNum];
+ final KubernetesStateHandleStore<TestingLongStateHandleHelper.LongStateHandle>[]
Review comment:
The Kubernetes-related ITCases are currently not running. I created [FLINK-20564](https://issues.apache.org/jira/browse/FLINK-22564) to cover this. -I haven't had the chance to run the ITCase locally, yet.- I checked that the test ran locally.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a change in pull request #15832: [FLINK-22494][ha] Introduce PossibleInconsistentStateException
Posted by GitBox <gi...@apache.org>.
XComp commented on a change in pull request #15832:
URL: https://github.com/apache/flink/pull/15832#discussion_r630810712
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperStateHandleStore.java
##########
@@ -196,29 +229,55 @@ public void replace(String pathInZooKeeper, IntegerResourceVersion expectedVersi
RetrievableStateHandle<T> newStateHandle = storage.store(state);
- boolean success = false;
+ final byte[] serializedStateHandle = serializeStateHandle(newStateHandle);
+ // initialize flags to serve the failure case
+ boolean discardOldState = false;
+ boolean discardNewState = true;
try {
- // Serialize the new state handle. This writes the state to the backend.
- byte[] serializedStateHandle = InstantiationUtil.serializeObject(newStateHandle);
+ setStateHandle(path, serializedStateHandle, expectedVersion.getValue());
+
+ // swap subject for deletion in case of success
+ discardOldState = true;
+ discardNewState = false;
+ } catch (Exception e) {
+ if (indicatesPossiblyInconsistentState(e)) {
+ // it's unclear whether the state handle metadata was written to ZooKeeper -
+ // hence, we don't discard any data
+ discardNewState = false;
+ throw new PossibleInconsistentStateException(e);
+ }
- // Replace state handle in ZooKeeper.
- client.setData()
- .withVersion(expectedVersion.getValue())
- .forPath(path, serializedStateHandle);
- success = true;
- } catch (KeeperException.NoNodeException e) {
// We wrap the exception here so that it could be caught in DefaultJobGraphStore
- throw new NotExistException("ZooKeeper node " + path + " does not exist.", e);
+ throw ExceptionUtils.findThrowable(e, KeeperException.NoNodeException.class)
+ .map(
+ nnee ->
+ new NotExistException(
+ "ZooKeeper node " + path + " does not exist.", nnee))
+ .orElseThrow(() -> e);
} finally {
- if (success) {
+ if (discardOldState) {
oldStateHandle.discardState();
- } else {
+ }
+
+ if (discardNewState) {
newStateHandle.discardState();
}
}
}
+ // this method is provided for the sole purpose of easier testing
+ @VisibleForTesting
+ protected void setStateHandle(String path, byte[] serializedStateHandle, int expectedVersion)
+ throws Exception {
+ // Replace state handle in ZooKeeper.
+ client.setData().withVersion(expectedVersion).forPath(path, serializedStateHandle);
+ }
+
+ private boolean indicatesPossiblyInconsistentState(Exception e) {
+ return !PRE_COMMIT_EXCEPTIONS.contains(e.getClass());
Review comment:
I would like to keep it more explicit. If subclasses are meant to be in that set, we should add them explicitly. Your example with `AuthExpiredException` is just a made up example, isn't it? I couldn't find it in the code. I was going through the error cases that are listed in [KeeperException.create](https://github.com/apache/zookeeper/blob/branch-3.4/zookeeper-server/src/main/java/org/apache/zookeeper/KeeperException.java#L93) to cover the possible cases that are triggered by the `commit()` call.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org