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/06/01 09:24:53 UTC
[1/4] flink git commit: [FLINK-1954] [FLINK-1955] [runtime] Set task
failure cause for produced result partitions
Repository: flink
Updated Branches:
refs/heads/master 5b799571d -> 93e960d8b
[FLINK-1954] [FLINK-1955] [runtime] Set task failure cause for produced result partitions
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/f75c16b0
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/f75c16b0
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/f75c16b0
Branch: refs/heads/master
Commit: f75c16b0540c839079188bb58b5acf2ede108767
Parents: dce1be1
Author: Ufuk Celebi <uc...@apache.org>
Authored: Thu May 28 14:58:31 2015 +0200
Committer: Ufuk Celebi <uc...@apache.org>
Committed: Mon Jun 1 09:21:55 2015 +0200
----------------------------------------------------------------------
.../runtime/io/network/NetworkEnvironment.java | 2 +-
.../io/network/netty/PartitionRequestQueue.java | 14 +++++
.../partition/PipelinedSubpartitionView.java | 5 ++
.../partition/ProducerFailedException.java | 32 +++++++++++
.../io/network/partition/ResultPartition.java | 17 +++++-
.../partition/ResultPartitionManager.java | 6 ++-
.../network/partition/ResultSubpartition.java | 4 ++
.../partition/ResultSubpartitionView.java | 2 +
.../partition/SpillableSubpartitionView.java | 5 ++
.../SpilledSubpartitionViewAsyncIO.java | 5 ++
.../SpilledSubpartitionViewSyncIO.java | 5 ++
.../partition/consumer/InputChannel.java | 4 ++
.../partition/consumer/LocalInputChannel.java | 12 ++++-
.../netty/CancelPartitionRequestTest.java | 5 ++
.../netty/PartitionRequestQueueTest.java | 56 ++++++++++++++++++++
.../consumer/LocalInputChannelTest.java | 28 ++++++++++
.../consumer/RemoteInputChannelTest.java | 27 +++++++++-
17 files changed, 223 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/f75c16b0/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
index 079a7f4..c082c6a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
@@ -342,7 +342,7 @@ public class NetworkEnvironment {
}
if (task.isCanceledOrFailed()) {
- partitionManager.releasePartitionsProducedBy(executionId);
+ partitionManager.releasePartitionsProducedBy(executionId, task.getFailureCause());
}
ResultPartitionWriter[] writers = task.getAllWriters();
http://git-wip-us.apache.org/repos/asf/flink/blob/f75c16b0/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
index c121091..d53a883 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
@@ -27,6 +27,7 @@ import io.netty.channel.ChannelInboundHandlerAdapter;
import org.apache.flink.runtime.io.network.api.EndOfPartitionEvent;
import org.apache.flink.runtime.io.network.api.serialization.EventSerializer;
import org.apache.flink.runtime.io.network.buffer.Buffer;
+import org.apache.flink.runtime.io.network.partition.ProducerFailedException;
import org.apache.flink.runtime.io.network.partition.ResultSubpartitionView;
import org.apache.flink.runtime.io.network.partition.consumer.InputChannelID;
import org.apache.flink.runtime.util.event.NotificationListener;
@@ -151,6 +152,14 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
else if (currentPartitionQueue.isReleased()) {
markAsReleased(currentPartitionQueue.getReceiverId());
+ Throwable cause = currentPartitionQueue.getFailureCause();
+
+ if (cause != null) {
+ ctx.writeAndFlush(new NettyMessage.ErrorResponse(
+ new ProducerFailedException(cause),
+ currentPartitionQueue.receiverId));
+ }
+
currentPartitionQueue = null;
}
}
@@ -292,6 +301,11 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
}
@Override
+ public Throwable getFailureCause() {
+ return queueIterator.getFailureCause();
+ }
+
+ @Override
public boolean registerListener(NotificationListener ignored) throws IOException {
return queueIterator.registerListener(this);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/f75c16b0/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionView.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionView.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionView.java
index 1e44324..51f1eab 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionView.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionView.java
@@ -70,4 +70,9 @@ class PipelinedSubpartitionView implements ResultSubpartitionView {
public boolean isReleased() {
return isReleased.get();
}
+
+ @Override
+ public Throwable getFailureCause() {
+ return parent.getFailureCause();
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/f75c16b0/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ProducerFailedException.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ProducerFailedException.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ProducerFailedException.java
new file mode 100644
index 0000000..60da409
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ProducerFailedException.java
@@ -0,0 +1,32 @@
+/*
+ * 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.network.partition;
+
+import org.apache.flink.runtime.execution.CancelTaskException;
+
+public class ProducerFailedException extends CancelTaskException {
+
+ private static final long serialVersionUID = -1555492656299526395L;
+
+ private final Throwable cause;
+
+ public ProducerFailedException(Throwable cause) {
+ this.cause = cause;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f75c16b0/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java
index bd9499e..98902f2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java
@@ -110,6 +110,8 @@ public class ResultPartition implements BufferPoolOwner {
private boolean isFinished;
+ private volatile Throwable cause;
+
// - Statistics ----------------------------------------------------------
/** The total number of buffers (both data and event buffers) */
@@ -275,13 +277,22 @@ public class ResultPartition implements BufferPoolOwner {
}
}
+ public void release() {
+ release(null);
+ }
+
/**
* Releases the result partition.
*/
- public void release() {
+ public void release(Throwable cause) {
if (isReleased.compareAndSet(false, true)) {
LOG.debug("{}: Releasing {}.", owningTaskName, this);
+ // Set the error cause
+ if (cause != null) {
+ this.cause = cause;
+ }
+
// Release all subpartitions
for (ResultSubpartition subpartition : subpartitions) {
try {
@@ -317,6 +328,10 @@ public class ResultPartition implements BufferPoolOwner {
return subpartitions[index].createReadView(bufferProvider);
}
+ public Throwable getFailureCause() {
+ return cause;
+ }
+
/**
* Releases buffers held by this result partition.
*
http://git-wip-us.apache.org/repos/asf/flink/blob/f75c16b0/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionManager.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionManager.java
index 5f25a4c..4063c1c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionManager.java
@@ -83,12 +83,16 @@ public class ResultPartitionManager implements ResultPartitionProvider {
}
public void releasePartitionsProducedBy(ExecutionAttemptID executionId) {
+ releasePartitionsProducedBy(executionId, null);
+ }
+
+ public void releasePartitionsProducedBy(ExecutionAttemptID executionId, Throwable cause) {
synchronized (registeredPartitions) {
final Map<IntermediateResultPartitionID, ResultPartition> partitions =
registeredPartitions.row(executionId);
for (ResultPartition partition : partitions.values()) {
- partition.release();
+ partition.release(cause);
}
for (IntermediateResultPartitionID partitionId : ImmutableList
http://git-wip-us.apache.org/repos/asf/flink/blob/f75c16b0/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartition.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartition.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartition.java
index e2ce16e..e9dfe32 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartition.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartition.java
@@ -67,6 +67,10 @@ public abstract class ResultSubpartition {
parent.onConsumedSubpartition(index);
}
+ protected Throwable getFailureCause() {
+ return parent.getFailureCause();
+ }
+
abstract public boolean add(Buffer buffer) throws IOException;
abstract public void finish() throws IOException;
http://git-wip-us.apache.org/repos/asf/flink/blob/f75c16b0/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartitionView.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartitionView.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartitionView.java
index dc445ba..cfc5455 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartitionView.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartitionView.java
@@ -55,4 +55,6 @@ public interface ResultSubpartitionView {
boolean isReleased();
+ Throwable getFailureCause();
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/f75c16b0/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionView.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionView.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionView.java
index 3d362d8..d37f042 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionView.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionView.java
@@ -160,4 +160,9 @@ class SpillableSubpartitionView implements ResultSubpartitionView {
public boolean isReleased() {
return isReleased.get();
}
+
+ @Override
+ public Throwable getFailureCause() {
+ return parent.getFailureCause();
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/f75c16b0/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpilledSubpartitionViewAsyncIO.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpilledSubpartitionViewAsyncIO.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpilledSubpartitionViewAsyncIO.java
index 1d4a9ab..ea5c20b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpilledSubpartitionViewAsyncIO.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpilledSubpartitionViewAsyncIO.java
@@ -190,6 +190,11 @@ class SpilledSubpartitionViewAsyncIO implements ResultSubpartitionView {
return isReleased;
}
+ @Override
+ public Throwable getFailureCause() {
+ return parent.getFailureCause();
+ }
+
/**
* Requests buffers from the buffer provider and triggers asynchronous read requests to fill
* them.
http://git-wip-us.apache.org/repos/asf/flink/blob/f75c16b0/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpilledSubpartitionViewSyncIO.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpilledSubpartitionViewSyncIO.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpilledSubpartitionViewSyncIO.java
index 62ebddf..24099a7 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpilledSubpartitionViewSyncIO.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpilledSubpartitionViewSyncIO.java
@@ -111,6 +111,11 @@ class SpilledSubpartitionViewSyncIO implements ResultSubpartitionView {
return isReleased.get();
}
+ @Override
+ public Throwable getFailureCause() {
+ return parent.getFailureCause();
+ }
+
/**
* A buffer pool to provide buffer to read the file into.
*
http://git-wip-us.apache.org/repos/asf/flink/blob/f75c16b0/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/InputChannel.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/InputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/InputChannel.java
index f828e2d..3998279 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/InputChannel.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/InputChannel.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.io.network.partition.consumer;
import org.apache.flink.runtime.event.task.TaskEvent;
+import org.apache.flink.runtime.execution.CancelTaskException;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.io.network.partition.ResultSubpartitionView;
@@ -156,6 +157,9 @@ public abstract class InputChannel {
final Throwable t = cause.get();
if (t != null) {
+ if (t instanceof CancelTaskException) {
+ throw (CancelTaskException) t;
+ }
if (t instanceof IOException) {
throw (IOException) t;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/f75c16b0/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
index 65f6a36..e72f612 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
@@ -24,6 +24,7 @@ import org.apache.flink.runtime.io.network.api.EndOfPartitionEvent;
import org.apache.flink.runtime.io.network.api.serialization.EventSerializer;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.partition.PartitionNotFoundException;
+import org.apache.flink.runtime.io.network.partition.ProducerFailedException;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.io.network.partition.ResultPartitionManager;
import org.apache.flink.runtime.io.network.partition.ResultSubpartitionView;
@@ -259,7 +260,16 @@ public class LocalInputChannel extends InputChannel implements NotificationListe
break;
}
- if (view.registerListener(this) || view.isReleased()) {
+ if (view.registerListener(this)) {
+ return;
+ }
+ else if (view.isReleased()) {
+ Throwable cause = view.getFailureCause();
+
+ if (cause != null) {
+ setError(new ProducerFailedException(cause));
+ }
+
return;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/f75c16b0/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CancelPartitionRequestTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CancelPartitionRequestTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CancelPartitionRequestTest.java
index 8f443a5..eb0d772 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CancelPartitionRequestTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CancelPartitionRequestTest.java
@@ -191,5 +191,10 @@ public class CancelPartitionRequestTest {
public boolean isReleased() {
return false;
}
+
+ @Override
+ public Throwable getFailureCause() {
+ return null;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/f75c16b0/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueueTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueueTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueueTest.java
new file mode 100644
index 0000000..594c686
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueueTest.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.io.network.netty;
+
+import io.netty.channel.embedded.EmbeddedChannel;
+import org.apache.flink.runtime.execution.CancelTaskException;
+import org.apache.flink.runtime.io.network.partition.ResultSubpartitionView;
+import org.apache.flink.runtime.io.network.partition.consumer.InputChannelID;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class PartitionRequestQueueTest {
+
+ @Test
+ public void testProducerFailedException() throws Exception {
+ PartitionRequestQueue queue = new PartitionRequestQueue();
+
+ EmbeddedChannel ch = new EmbeddedChannel(queue);
+
+ ResultSubpartitionView view = mock(ResultSubpartitionView.class);
+ when(view.isReleased()).thenReturn(true);
+ when(view.getFailureCause()).thenReturn(new RuntimeException("Expected test exception"));
+
+ // Enqueue the erroneous view
+ queue.enqueue(view, new InputChannelID());
+ ch.runPendingTasks();
+
+ // Read the enqueued msg
+ Object msg = ch.readOutbound();
+
+ assertEquals(msg.getClass(), NettyMessage.ErrorResponse.class);
+
+ NettyMessage.ErrorResponse err = (NettyMessage.ErrorResponse) msg;
+ assertTrue(err.error instanceof CancelTaskException);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f75c16b0/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannelTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannelTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannelTest.java
index c25e1d8..938c2ad 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannelTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannelTest.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.io.network.partition.consumer;
import com.google.common.collect.Lists;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.execution.CancelTaskException;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.io.network.ConnectionID;
@@ -37,6 +38,7 @@ import org.apache.flink.runtime.io.network.partition.ResultPartitionConsumableNo
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.io.network.partition.ResultPartitionManager;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
+import org.apache.flink.runtime.io.network.partition.ResultSubpartitionView;
import org.apache.flink.runtime.io.network.util.TestBufferFactory;
import org.apache.flink.runtime.io.network.util.TestPartitionProducer;
import org.apache.flink.runtime.io.network.util.TestProducerSource;
@@ -62,6 +64,7 @@ import static com.google.common.base.Preconditions.checkArgument;
import static org.apache.flink.runtime.io.disk.iomanager.IOManager.IOMode.ASYNC;
import static org.junit.Assert.fail;
import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyInt;
import static org.mockito.Matchers.anyLong;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.doAnswer;
@@ -230,6 +233,31 @@ public class LocalInputChannelTest {
}
}
+ @Test(expected = CancelTaskException.class)
+ public void testProducerFailedException() throws Exception {
+
+ ResultSubpartitionView view = mock(ResultSubpartitionView.class);
+ when(view.isReleased()).thenReturn(true);
+ when(view.getFailureCause()).thenReturn(new Exception("Expected test exception"));
+
+ ResultPartitionManager partitionManager = mock(ResultPartitionManager.class);
+ when(partitionManager
+ .createSubpartitionView(any(ResultPartitionID.class), anyInt(), any(BufferProvider.class)))
+ .thenReturn(view);
+
+ SingleInputGate inputGate = mock(SingleInputGate.class);
+ BufferProvider bufferProvider = mock(BufferProvider.class);
+ when(inputGate.getBufferProvider()).thenReturn(bufferProvider);
+
+ LocalInputChannel ch = createLocalInputChannel(
+ inputGate, partitionManager, new Tuple2<Integer, Integer>(0, 0));
+
+ ch.requestSubpartition(0);
+
+ // Should throw an instance of CancelTaskException.
+ ch.getNextBuffer();
+ }
+
// ---------------------------------------------------------------------------------------------
private LocalInputChannel createLocalInputChannel(
http://git-wip-us.apache.org/repos/asf/flink/blob/f75c16b0/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java
index 22a11f9..c484cc4 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java
@@ -19,9 +19,11 @@
package org.apache.flink.runtime.io.network.partition.consumer;
import com.google.common.collect.Lists;
+import org.apache.flink.runtime.execution.CancelTaskException;
import org.apache.flink.runtime.io.network.ConnectionID;
import org.apache.flink.runtime.io.network.ConnectionManager;
import org.apache.flink.runtime.io.network.netty.PartitionRequestClient;
+import org.apache.flink.runtime.io.network.partition.ProducerFailedException;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.io.network.util.TestBufferFactory;
import org.junit.Test;
@@ -245,14 +247,35 @@ public class RemoteInputChannelTest {
0,
partitionId,
mock(ConnectionID.class),
- connectionManager
- );
+ connectionManager);
ch.onFailedPartitionRequest();
verify(inputGate).triggerPartitionStateCheck(eq(partitionId));
}
+ @Test(expected = CancelTaskException.class)
+ public void testProducerFailedException() throws Exception {
+
+ ConnectionManager connManager = mock(ConnectionManager.class);
+ when(connManager.createPartitionRequestClient(any(ConnectionID.class)))
+ .thenReturn(mock(PartitionRequestClient.class));
+
+ final RemoteInputChannel ch = new RemoteInputChannel(
+ mock(SingleInputGate.class),
+ 0,
+ new ResultPartitionID(),
+ mock(ConnectionID.class),
+ connManager);
+
+ ch.onError(new ProducerFailedException(new RuntimeException("Expected test exception.")));
+
+ ch.requestSubpartition(0);
+
+ // Should throw an instance of CancelTaskException.
+ ch.getNextBuffer();
+ }
+
// ---------------------------------------------------------------------------------------------
private RemoteInputChannel createRemoteInputChannel(SingleInputGate inputGate)
[4/4] flink git commit: [runtime] Rename internal field of Netty
error message
Posted by uc...@apache.org.
[runtime] Rename internal field of Netty error message
This closes #746.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/93e960d8
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/93e960d8
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/93e960d8
Branch: refs/heads/master
Commit: 93e960d8ba2a52bd070609fd6ded86b2731596eb
Parents: 6182180
Author: Ufuk Celebi <uc...@apache.org>
Authored: Fri May 29 10:54:16 2015 +0200
Committer: Ufuk Celebi <uc...@apache.org>
Committed: Mon Jun 1 09:24:17 2015 +0200
----------------------------------------------------------------------
.../flink/runtime/io/network/netty/NettyMessage.java | 14 +++++++-------
.../network/netty/PartitionRequestClientHandler.java | 6 +++---
.../network/netty/NettyMessageSerializationTest.java | 12 ++++++------
.../io/network/netty/PartitionRequestQueueTest.java | 2 +-
4 files changed, 17 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/93e960d8/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
index d0840b0..53afd03 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
@@ -264,19 +264,19 @@ abstract class NettyMessage {
private static final byte ID = 1;
- Throwable error;
+ Throwable cause;
InputChannelID receiverId;
public ErrorResponse() {
}
- ErrorResponse(Throwable error) {
- this.error = error;
+ ErrorResponse(Throwable cause) {
+ this.cause = cause;
}
- ErrorResponse(Throwable error, InputChannelID receiverId) {
- this.error = error;
+ ErrorResponse(Throwable cause, InputChannelID receiverId) {
+ this.cause = cause;
this.receiverId = receiverId;
}
@@ -297,7 +297,7 @@ abstract class NettyMessage {
oos = new ObjectOutputStream(new DataOutputViewStream(outputView));
- oos.writeObject(error);
+ oos.writeObject(cause);
if (receiverId != null) {
result.writeBoolean(true);
@@ -338,7 +338,7 @@ abstract class NettyMessage {
throw new ClassCastException("Read object expected to be of type Throwable, " +
"actual type is " + obj.getClass() + ".");
} else {
- error = (Throwable) obj;
+ cause = (Throwable) obj;
if (buffer.readBoolean()) {
receiverId = InputChannelID.fromByteBuf(buffer);
http://git-wip-us.apache.org/repos/asf/flink/blob/93e960d8/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
index 985e2a5..508cac9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
@@ -222,19 +222,19 @@ class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
if (error.isFatalError()) {
notifyAllChannelsOfErrorAndClose(new RemoteTransportException(
"Fatal error at remote task manager '" + remoteAddr + "'.",
- remoteAddr, error.error));
+ remoteAddr, error.cause));
}
else {
RemoteInputChannel inputChannel = inputChannels.get(error.receiverId);
if (inputChannel != null) {
- if (error.error.getClass() == PartitionNotFoundException.class) {
+ if (error.cause.getClass() == PartitionNotFoundException.class) {
inputChannel.onFailedPartitionRequest();
}
else {
inputChannel.onError(new RemoteTransportException(
"Error at remote task manager '" + remoteAddr + "'.",
- remoteAddr, error.error));
+ remoteAddr, error.cause));
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/93e960d8/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyMessageSerializationTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyMessageSerializationTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyMessageSerializationTest.java
index b464692..b1315be 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyMessageSerializationTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyMessageSerializationTest.java
@@ -93,8 +93,8 @@ public class NettyMessageSerializationTest {
NettyMessage.ErrorResponse expected = new NettyMessage.ErrorResponse(expectedError, receiverId);
NettyMessage.ErrorResponse actual = encodeAndDecode(expected);
- assertEquals(expected.error.getClass(), actual.error.getClass());
- assertEquals(expected.error.getMessage(), actual.error.getMessage());
+ assertEquals(expected.cause.getClass(), actual.cause.getClass());
+ assertEquals(expected.cause.getMessage(), actual.cause.getMessage());
assertEquals(receiverId, actual.receiverId);
}
@@ -105,8 +105,8 @@ public class NettyMessageSerializationTest {
NettyMessage.ErrorResponse expected = new NettyMessage.ErrorResponse(expectedError, receiverId);
NettyMessage.ErrorResponse actual = encodeAndDecode(expected);
- assertEquals(expected.error.getClass(), actual.error.getClass());
- assertEquals(expected.error.getMessage(), actual.error.getMessage());
+ assertEquals(expected.cause.getClass(), actual.cause.getClass());
+ assertEquals(expected.cause.getMessage(), actual.cause.getMessage());
assertEquals(receiverId, actual.receiverId);
}
@@ -116,8 +116,8 @@ public class NettyMessageSerializationTest {
NettyMessage.ErrorResponse expected = new NettyMessage.ErrorResponse(expectedError);
NettyMessage.ErrorResponse actual = encodeAndDecode(expected);
- assertEquals(expected.error.getClass(), actual.error.getClass());
- assertEquals(expected.error.getMessage(), actual.error.getMessage());
+ assertEquals(expected.cause.getClass(), actual.cause.getClass());
+ assertEquals(expected.cause.getMessage(), actual.cause.getMessage());
assertNull(actual.receiverId);
assertTrue(actual.isFatalError());
}
http://git-wip-us.apache.org/repos/asf/flink/blob/93e960d8/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueueTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueueTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueueTest.java
index 594c686..3f281bd 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueueTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueueTest.java
@@ -51,6 +51,6 @@ public class PartitionRequestQueueTest {
assertEquals(msg.getClass(), NettyMessage.ErrorResponse.class);
NettyMessage.ErrorResponse err = (NettyMessage.ErrorResponse) msg;
- assertTrue(err.error instanceof CancelTaskException);
+ assertTrue(err.cause instanceof CancelTaskException);
}
}
[3/4] flink git commit: [runtime] Fix uncaught exception in Netty
server pipeline
Posted by uc...@apache.org.
[runtime] Fix uncaught exception in Netty server pipeline
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/61821801
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/61821801
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/61821801
Branch: refs/heads/master
Commit: 61821801878ed7bcebbcc98405ba5b13be1cbc2a
Parents: f75c16b
Author: Ufuk Celebi <uc...@apache.org>
Authored: Thu May 28 15:57:39 2015 +0200
Committer: Ufuk Celebi <uc...@apache.org>
Committed: Mon Jun 1 09:21:56 2015 +0200
----------------------------------------------------------------------
.../io/network/netty/PartitionRequestQueue.java | 19 ++++++++++++-------
1 file changed, 12 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/61821801/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
index d53a883..5301195 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
@@ -242,14 +242,19 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
- if (future.isSuccess()) {
- writeAndFlushNextMessageIfPossible(future.channel());
- }
- else if (future.cause() != null) {
- handleException(future.channel(), future.cause());
+ try {
+ if (future.isSuccess()) {
+ writeAndFlushNextMessageIfPossible(future.channel());
+ }
+ else if (future.cause() != null) {
+ handleException(future.channel(), future.cause());
+ }
+ else {
+ handleException(future.channel(), new IllegalStateException("Sending cancelled by user."));
+ }
}
- else {
- handleException(future.channel(), new IllegalStateException("Sending cancelled by user."));
+ catch (Throwable t) {
+ handleException(future.channel(), t);
}
}
}
[2/4] flink git commit: [FLINK-1954] [FLINK-1958] [runtime] Cancel
transfers of failed receiver tasks
Posted by uc...@apache.org.
[FLINK-1954] [FLINK-1958] [runtime] Cancel transfers of failed receiver tasks
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/dce1be18
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/dce1be18
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/dce1be18
Branch: refs/heads/master
Commit: dce1be18593539ff29c3d55c5f2c1208a2e54c10
Parents: 5b79957
Author: Ufuk Celebi <uc...@apache.org>
Authored: Thu May 28 11:45:52 2015 +0200
Committer: Ufuk Celebi <uc...@apache.org>
Committed: Mon Jun 1 09:21:55 2015 +0200
----------------------------------------------------------------------
.../network/netty/PartitionRequestClient.java | 10 ++-
.../netty/PartitionRequestClientHandler.java | 23 +++----
.../io/network/netty/PartitionRequestQueue.java | 27 +++++++-
.../partition/consumer/RemoteInputChannel.java | 6 +-
.../netty/CancelPartitionRequestTest.java | 68 +++++++++++++++++++-
5 files changed, 111 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/dce1be18/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClient.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClient.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClient.java
index cb4ecb6..3049af6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClient.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClient.java
@@ -58,7 +58,12 @@ public class PartitionRequestClient {
// If zero, the underlying TCP channel can be safely closed
private final AtomicDisposableReferenceCounter closeReferenceCounter = new AtomicDisposableReferenceCounter();
- PartitionRequestClient(Channel tcpChannel, PartitionRequestClientHandler partitionRequestHandler, ConnectionID connectionId, PartitionRequestClientFactory clientFactory) {
+ PartitionRequestClient(
+ Channel tcpChannel,
+ PartitionRequestClientHandler partitionRequestHandler,
+ ConnectionID connectionId,
+ PartitionRequestClientFactory clientFactory) {
+
this.tcpChannel = checkNotNull(tcpChannel);
this.partitionRequestHandler = checkNotNull(partitionRequestHandler);
this.connectionId = checkNotNull(connectionId);
@@ -168,5 +173,8 @@ public class PartitionRequestClient {
// Make sure to remove the client from the factory
clientFactory.destroyPartitionRequestClient(connectionId, this);
}
+ else {
+ partitionRequestHandler.cancelRequestFor(inputChannel.getInputChannelId());
+ }
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/dce1be18/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
index 5964b49..985e2a5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
@@ -18,7 +18,7 @@
package org.apache.flink.runtime.io.network.netty;
-import com.google.common.collect.Sets;
+import com.google.common.collect.Maps;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.apache.flink.core.memory.MemorySegment;
@@ -39,7 +39,6 @@ import java.io.IOException;
import java.net.SocketAddress;
import java.util.ArrayDeque;
import java.util.Queue;
-import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -65,7 +64,7 @@ class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
* Set of cancelled partition requests. A request is cancelled iff an input channel is cleared
* while data is still coming in for this channel.
*/
- private volatile Set<InputChannelID> cancelled;
+ private final ConcurrentMap<InputChannelID, InputChannelID> cancelled = Maps.newConcurrentMap();
private ChannelHandlerContext ctx;
@@ -85,6 +84,12 @@ class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
inputChannels.remove(listener.getInputChannelId());
}
+ void cancelRequestFor(InputChannelID inputChannelId) {
+ if (cancelled.putIfAbsent(inputChannelId, inputChannelId) == null) {
+ ctx.writeAndFlush(new NettyMessage.CancelPartitionRequest(inputChannelId));
+ }
+ }
+
// ------------------------------------------------------------------------
// Network events
// ------------------------------------------------------------------------
@@ -183,18 +188,6 @@ class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
}
}
- private void cancelRequestFor(InputChannelID inputChannelId) {
- if (cancelled == null) {
- cancelled = Sets.newConcurrentHashSet();
- }
-
- if (!cancelled.contains(inputChannelId)) {
- ctx.writeAndFlush(new NettyMessage.CancelPartitionRequest(inputChannelId));
-
- cancelled.add(inputChannelId);
- }
- }
-
// ------------------------------------------------------------------------
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/dce1be18/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
index 279ddbe..c121091 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.io.network.netty;
+import com.google.common.collect.Sets;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
@@ -35,6 +36,7 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.ArrayDeque;
import java.util.Queue;
+import java.util.Set;
import static com.google.common.base.Preconditions.checkNotNull;
import static org.apache.flink.runtime.io.network.netty.NettyMessage.BufferResponse;
@@ -51,6 +53,8 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
private final Queue<SequenceNumberingSubpartitionView> queue = new ArrayDeque<SequenceNumberingSubpartitionView>();
+ private final Set<InputChannelID> released = Sets.newHashSet();
+
private SequenceNumberingSubpartitionView currentPartitionQueue;
private boolean fatalError;
@@ -88,9 +92,14 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
else if (msg.getClass() == InputChannelID.class) {
InputChannelID toCancel = (InputChannelID) msg;
+ if (released.contains(toCancel)) {
+ return;
+ }
+
// Cancel the request for the input channel
if (currentPartitionQueue != null && currentPartitionQueue.getReceiverId().equals(toCancel)) {
currentPartitionQueue.releaseAllResources();
+ markAsReleased(currentPartitionQueue.receiverId);
currentPartitionQueue = null;
}
else {
@@ -101,6 +110,7 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
if (curr.getReceiverId().equals(toCancel)) {
curr.releaseAllResources();
+ markAsReleased(curr.receiverId);
}
else {
queue.add(curr);
@@ -139,6 +149,8 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
currentPartitionQueue = null;
}
else if (currentPartitionQueue.isReleased()) {
+ markAsReleased(currentPartitionQueue.getReceiverId());
+
currentPartitionQueue = null;
}
}
@@ -150,8 +162,9 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
currentPartitionQueue.notifySubpartitionConsumed();
currentPartitionQueue.releaseAllResources();
- currentPartitionQueue = null;
+ markAsReleased(currentPartitionQueue.getReceiverId());
+ currentPartitionQueue = null;
}
channel.writeAndFlush(resp).addListener(writeListener);
@@ -194,14 +207,25 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
private void releaseAllResources() throws IOException {
if (currentPartitionQueue != null) {
currentPartitionQueue.releaseAllResources();
+ markAsReleased(currentPartitionQueue.getReceiverId());
+
currentPartitionQueue = null;
}
while ((currentPartitionQueue = queue.poll()) != null) {
currentPartitionQueue.releaseAllResources();
+
+ markAsReleased(currentPartitionQueue.getReceiverId());
}
}
+ /**
+ * Marks a receiver as released.
+ */
+ private void markAsReleased(InputChannelID receiverId) {
+ released.add(receiverId);
+ }
+
// This listener is called after an element of the current queue has been
// flushed. If successful, the listener triggers further processing of the
// queues.
@@ -285,5 +309,4 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
ctx.pipeline().fireUserEventTriggered(this);
}
}
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/dce1be18/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
index 090e94d..b70c3a8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
@@ -75,7 +75,7 @@ public class RemoteInputChannel extends InputChannel {
*/
private int expectedSequenceNumber = 0;
- RemoteInputChannel(
+ public RemoteInputChannel(
SingleInputGate inputGate,
int channelIndex,
ResultPartitionID partitionId,
@@ -86,7 +86,7 @@ public class RemoteInputChannel extends InputChannel {
new Tuple2<Integer, Integer>(0, 0));
}
- RemoteInputChannel(
+ public RemoteInputChannel(
SingleInputGate inputGate,
int channelIndex,
ResultPartitionID partitionId,
@@ -193,6 +193,8 @@ public class RemoteInputChannel extends InputChannel {
}
}
+ // The released flag has to be set before closing the connection to ensure that
+ // buffers received concurrently with closing are properly recycled.
if (partitionRequestClient != null) {
partitionRequestClient.close(this);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/dce1be18/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CancelPartitionRequestTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CancelPartitionRequestTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CancelPartitionRequestTest.java
index 75f4284..8f443a5 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CancelPartitionRequestTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CancelPartitionRequestTest.java
@@ -38,7 +38,8 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import static com.google.common.base.Preconditions.checkNotNull;
-import static org.apache.flink.runtime.io.network.netty.NettyMessage.*;
+import static org.apache.flink.runtime.io.network.netty.NettyMessage.CancelPartitionRequest;
+import static org.apache.flink.runtime.io.network.netty.NettyMessage.PartitionRequest;
import static org.apache.flink.runtime.io.network.netty.NettyTestUtil.connect;
import static org.apache.flink.runtime.io.network.netty.NettyTestUtil.initServerAndClient;
import static org.apache.flink.runtime.io.network.netty.NettyTestUtil.shutdown;
@@ -46,13 +47,17 @@ import static org.junit.Assert.fail;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
public class CancelPartitionRequestTest {
/**
* Verifies that requests for non-existing (failed/cancelled) input channels are properly
- * cancelled.
+ * cancelled. The receiver receives data, but there is no input channel to receive the data.
+ * This should cancel the request.
*/
@Test
public void testCancelPartitionRequest() throws Exception {
@@ -68,9 +73,11 @@ public class CancelPartitionRequestTest {
CountDownLatch sync = new CountDownLatch(1);
+ ResultSubpartitionView view = spy(new InfiniteSubpartitionView(outboundBuffers, sync));
+
// Return infinite subpartition
when(partitions.createSubpartitionView(eq(pid), eq(0), any(BufferProvider.class)))
- .thenReturn(new InfiniteSubpartitionView(outboundBuffers, sync));
+ .thenReturn(view);
PartitionRequestProtocol protocol = new PartitionRequestProtocol(
partitions, mock(TaskEventDispatcher.class), mock(NetworkBufferPool.class));
@@ -87,6 +94,61 @@ public class CancelPartitionRequestTest {
fail("Timed out after waiting for " + TestingUtils.TESTING_DURATION().toMillis() +
" ms to be notified about cancelled partition.");
}
+
+ verify(view, times(1)).releaseAllResources();
+ verify(view, times(0)).notifySubpartitionConsumed();
+ }
+ finally {
+ shutdown(serverAndClient);
+ }
+ }
+
+ @Test
+ public void testDuplicateCancel() throws Exception {
+
+ NettyServerAndClient serverAndClient = null;
+
+ try {
+ TestPooledBufferProvider outboundBuffers = new TestPooledBufferProvider(16);
+
+ ResultPartitionManager partitions = mock(ResultPartitionManager.class);
+
+ ResultPartitionID pid = new ResultPartitionID();
+
+ CountDownLatch sync = new CountDownLatch(1);
+
+ ResultSubpartitionView view = spy(new InfiniteSubpartitionView(outboundBuffers, sync));
+
+ // Return infinite subpartition
+ when(partitions.createSubpartitionView(eq(pid), eq(0), any(BufferProvider.class)))
+ .thenReturn(view);
+
+ PartitionRequestProtocol protocol = new PartitionRequestProtocol(
+ partitions, mock(TaskEventDispatcher.class), mock(NetworkBufferPool.class));
+
+ serverAndClient = initServerAndClient(protocol);
+
+ Channel ch = connect(serverAndClient);
+
+ // Request for non-existing input channel => results in cancel request
+ InputChannelID inputChannelId = new InputChannelID();
+
+ ch.writeAndFlush(new PartitionRequest(pid, 0, inputChannelId)).await();
+
+ // Wait for the notification
+ if (!sync.await(TestingUtils.TESTING_DURATION().toMillis(), TimeUnit.MILLISECONDS)) {
+ fail("Timed out after waiting for " + TestingUtils.TESTING_DURATION().toMillis() +
+ " ms to be notified about cancelled partition.");
+ }
+
+ ch.writeAndFlush(new CancelPartitionRequest(inputChannelId)).await();
+
+ ch.close();
+
+ NettyTestUtil.awaitClose(ch);
+
+ verify(view, times(1)).releaseAllResources();
+ verify(view, times(0)).notifySubpartitionConsumed();
}
finally {
shutdown(serverAndClient);