Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/1017#discussion_r37168797
--- Diff:
flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
---
@@ -39,61 +37,40 @@
*/
public class SourceStreamTask<OUT> extends StreamTask<OUT,
StreamSource<OUT>> {
- private static final Logger LOG =
LoggerFactory.getLogger(SourceStreamTask.class);
-
@Override
- public void invoke() throws Exception {
- final SourceOutput<StreamRecord<OUT>> output = new
SourceOutput<StreamRecord<OUT>>(outputHandler.getOutput(), checkpointLock);
-
- boolean operatorOpen = false;
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("Task {} invoked", getName());
- }
-
- try {
- openOperator();
- operatorOpen = true;
-
- streamOperator.run(checkpointLock, output);
-
- closeOperator();
- operatorOpen = false;
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("Task {} invocation finished",
getName());
- }
-
- }
- catch (Exception e) {
- LOG.error(getEnvironment().getTaskNameWithSubtasks() +
" failed", e);
-
- if (operatorOpen) {
- try {
- closeOperator();
- }
- catch (Throwable t) {
- LOG.warn("Exception while closing
operator.", t);
- }
- }
- throw e;
- }
- finally {
- this.isRunning = false;
- // Cleanup
- outputHandler.flushOutputs();
- clearBuffers();
- }
+ protected void init() {
+ // does not hold any resources, so no initialization needed
+ }
+ @Override
+ protected void cleanup() {
+ // does not hold any resources, so no cleanup needed
}
+
@Override
- public void cancel() {
- super.cancel();
+ protected void run() throws Exception {
+ final Object checkpointLock = getCheckpointLock();
+
+ final SourceOutput<StreamRecord<OUT>> output =
+ new
SourceOutput<StreamRecord<OUT>>(outputHandler.getOutput(), checkpointLock);
+
+ streamOperator.run(checkpointLock, output);
+ }
+
+ @Override
+ protected void cancelTask() throws Exception {
streamOperator.cancel();
}
+ //
------------------------------------------------------------------------
+
+ // TODO:
+ // does this help with anything? The losk should be already held by the
source function that
+ // emits. If that one does not hold the lock, then this does not help
either.
+
--- End diff --
This is for the case where the source does not actually acquire the lock
(because it is not participating in checkpointing). This might be a degenerate
case, however, and not really necessary.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
