You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by StefanRRichter <gi...@git.apache.org> on 2017/09/14 09:45:48 UTC
[GitHub] flink pull request #4671: [FLINK-7619] Improve abstraction in AbstractAsyncI...
GitHub user StefanRRichter opened a pull request:
https://github.com/apache/flink/pull/4671
[FLINK-7619] Improve abstraction in AbstractAsyncIOCallable to better fit
## What is the purpose of the change
This PR improves the abstractions around resource management for async backend snapshots.
## Brief change log
- return value when removing streams from {{CloseableRegistry}} to not rely in idempotent close, but enforce close to be called exactly once.
- Introduces {{AbstractAsyncCallableWithResources}} as a more general abstraction to manage resources in async checkpointing.
## Verifying this change
This change is already covered by existing tests.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes)
- 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, Yarn/Mesos, ZooKeeper: (yes)
## Documentation
- Does this pull request introduce a new feature? (no)
- If yes, how is the feature documented? (not applicable)
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/StefanRRichter/flink improve-snapshot-resourcemngmt
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/4671.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #4671
----
----
---
[GitHub] flink issue #4671: [FLINK-7619] Improve abstraction in AbstractAsyncIOCallab...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on the issue:
https://github.com/apache/flink/pull/4671
@aljoscha Thanks for the reviews of this and PR #4639 ! I addressed your comments and will proceed to merge both.
---
[GitHub] flink issue #4671: [FLINK-7619] Improve abstraction in AbstractAsyncIOCallab...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on the issue:
https://github.com/apache/flink/pull/4671
CC @aljoscha
---
[GitHub] flink issue #4671: [FLINK-7619] Improve abstraction in AbstractAsyncIOCallab...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on the issue:
https://github.com/apache/flink/pull/4671
Merged in 5af463a9c0.
---
[GitHub] flink pull request #4671: [FLINK-7619] Improve abstraction in AbstractAsyncI...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/4671#discussion_r140458205
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/io/async/AbstractAsyncCallableWithResources.java ---
@@ -0,0 +1,194 @@
+/*
+ * 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.io.async;
+
+import org.apache.flink.util.ExceptionUtils;
+
+import java.io.IOException;
+
+/**
+ * The abstract class encapsulates the lifecycle and execution strategy for asynchronous operations that use resources.
+ *
+ * @param <V> return type of the asynchronous call.
+ */
+public abstract class AbstractAsyncCallableWithResources<V> implements StoppableCallbackCallable<V> {
+
+ /** Tracks if the stop method was called on this object. */
+ private volatile boolean stopped;
+
+ /** Tracks if call method was executed (only before stop calls). */
+ private volatile boolean called;
+
+ /** Stores a collected exception if there was on during stop. */
--- End diff --
nit: "if there was one ..."
---
[GitHub] flink pull request #4671: [FLINK-7619] Improve abstraction in AbstractAsyncI...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/4671#discussion_r140458159
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/io/async/AbstractAsyncCallableWithResources.java ---
@@ -0,0 +1,194 @@
+/*
+ * 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.io.async;
+
+import org.apache.flink.util.ExceptionUtils;
+
+import java.io.IOException;
+
+/**
+ * The abstract class encapsulates the lifecycle and execution strategy for asynchronous operations that use resources.
--- End diff --
nit: "This class..."
---
[GitHub] flink pull request #4671: [FLINK-7619] Improve abstraction in AbstractAsyncI...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/4671#discussion_r140456968
--- Diff: flink-core/src/main/java/org/apache/flink/util/AbstractCloseableRegistry.java ---
@@ -90,14 +90,14 @@ public final void registerClosable(C closeable) throws IOException {
*
* @param closeable instance to remove from the registry.
*/
- public final void unregisterClosable(C closeable) {
+ public final boolean unregisterCloseable(C closeable) {
--- End diff --
Please add a description of the return parameter. Doesn't need to be a `@param` block, a description in the text is enough. Same for the method below.
---
[GitHub] flink pull request #4671: [FLINK-7619] Improve abstraction in AbstractAsyncI...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter closed the pull request at:
https://github.com/apache/flink/pull/4671
---
[GitHub] flink pull request #4671: [FLINK-7619] Improve abstraction in AbstractAsyncI...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/4671#discussion_r140458424
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/io/async/AbstractAsyncCallableWithResources.java ---
@@ -0,0 +1,194 @@
+/*
+ * 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.io.async;
+
+import org.apache.flink.util.ExceptionUtils;
+
+import java.io.IOException;
+
+/**
+ * The abstract class encapsulates the lifecycle and execution strategy for asynchronous operations that use resources.
+ *
+ * @param <V> return type of the asynchronous call.
+ */
+public abstract class AbstractAsyncCallableWithResources<V> implements StoppableCallbackCallable<V> {
+
+ /** Tracks if the stop method was called on this object. */
+ private volatile boolean stopped;
+
+ /** Tracks if call method was executed (only before stop calls). */
+ private volatile boolean called;
+
+ /** Stores a collected exception if there was on during stop. */
+ private volatile Exception stopException;
+
+ public AbstractAsyncCallableWithResources() {
+ this.stopped = false;
+ this.called = false;
+ }
+
+ /**
+ * This method implements the strategy for the actual IO operation:
+ * <p>
+ * 1) Acquire resources asynchronously and atomically w.r.t stopping.
+ * 2) Performs the operation
+ * 3) Releases resources.
+ *
+ * @return Result of the IO operation, e.g. a deserialized object.
+ * @throws Exception exception that happened during the call.
+ */
+ @Override
+ public V call() throws Exception {
--- End diff --
Should this be final?
---