You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by uc...@apache.org on 2015/05/26 10:34:30 UTC
flink git commit: [runtime] Fix possible IndexOutOfBoundsException
when scheduling or updating consumers
Repository: flink
Updated Branches:
refs/heads/master 4dcfc43b8 -> 44c247bda
[runtime] Fix possible IndexOutOfBoundsException when scheduling or updating consumers
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/44c247bd
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/44c247bd
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/44c247bd
Branch: refs/heads/master
Commit: 44c247bda9581c02e967841468e8f715df3cb6e6
Parents: 4dcfc43
Author: Ufuk Celebi <uc...@apache.org>
Authored: Tue May 26 10:33:37 2015 +0200
Committer: Ufuk Celebi <uc...@apache.org>
Committed: Tue May 26 10:34:13 2015 +0200
----------------------------------------------------------------------
.../org/apache/flink/runtime/executiongraph/Execution.java | 7 ++++++-
1 file changed, 6 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/44c247bd/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
index eae696c..7972d43 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
@@ -433,9 +433,14 @@ public class Execution implements Serializable {
}
void scheduleOrUpdateConsumers(List<List<ExecutionEdge>> allConsumers) {
- if (allConsumers.size() > 1) {
+ final int numConsumers = allConsumers.size();
+
+ if (numConsumers > 1) {
fail(new IllegalStateException("Currently, only a single consumer group per partition is supported."));
}
+ else if (numConsumers == 0) {
+ return;
+ }
for (ExecutionEdge edge : allConsumers.get(0)) {
final ExecutionVertex consumerVertex = edge.getTarget();