[flink-web] branch asf-site updated: Add zhijiang to community page
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a commit to branch asf-site in repository https://gitbox.apache.org/repos/asf/flink-web.git The following commit(s) were added to refs/heads/asf-site by this push: new a0c3fab Add zhijiang to community page a0c3fab is described below commit a0c3faba64a66bb5c17889f7d893e4b17de6321d Author: zhijiang AuthorDate: Tue Jul 23 14:22:03 2019 +0800 Add zhijiang to community page --- community.md | 6 ++ community.zh.md| 6 ++ content/community.html | 6 ++ 3 files changed, 18 insertions(+) diff --git a/community.md b/community.md index 6443f19..4fd0b12 100644 --- a/community.md +++ b/community.md @@ -395,6 +395,12 @@ Flink Forward is a conference happening yearly in different locations around the shaoxuan +https://avatars3.githubusercontent.com/u/12387855?s=50"; class="committer-avatar"> +Zhijiang Wang + Committer +zhijiang + + https://avatars1.githubusercontent.com/u/1826769?s=50"; class="committer-avatar"> Daniel Warneke PMC, Committer diff --git a/community.zh.md b/community.zh.md index 9b9c6a3..b596f1c 100644 --- a/community.zh.md +++ b/community.zh.md @@ -390,6 +390,12 @@ Flink Forward 大会每年都会在世界的不同地方举办。关于大会最 shaoxuan +https://avatars3.githubusercontent.com/u/12387855?s=50"; class="committer-avatar"> +Zhijiang Wang +Committer +zhijiang + + https://avatars1.githubusercontent.com/u/1826769?s=50"; class="committer-avatar"> Daniel Warneke PMC, Committer diff --git a/content/community.html b/content/community.html index ad08e83..5b988de 100644 --- a/content/community.html +++ b/content/community.html @@ -583,6 +583,12 @@ shaoxuan +https://avatars3.githubusercontent.com/u/12387855?s=50"; class="committer-avatar" /> +Zhijiang Wang +Committer +zhijiang + + https://avatars1.githubusercontent.com/u/1826769?s=50"; class="committer-avatar" /> Daniel Warneke PMC, Committer
[flink] 11/13: [hotfix][tests] fix codestyle issues in NettyShuffleDescriptorBuilder
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 8735a5bc0de4fbe924ae6f9a4c1cc62459206570
Author: Andrey Zagrebin
AuthorDate: Mon Jul 29 17:49:24 2019 +0300
[hotfix][tests] fix codestyle issues in NettyShuffleDescriptorBuilder
---
.../org/apache/flink/runtime/util/NettyShuffleDescriptorBuilder.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/util/NettyShuffleDescriptorBuilder.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/util/NettyShuffleDescriptorBuilder.java
index 2d58d03..052fefb 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/util/NettyShuffleDescriptorBuilder.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/util/NettyShuffleDescriptorBuilder.java
@@ -38,8 +38,8 @@ public class NettyShuffleDescriptorBuilder {
private ResourceID producerLocation = ResourceID.generate();
private ResultPartitionID id = new ResultPartitionID();
private InetAddress address = InetAddress.getLoopbackAddress();
- private int dataPort = 0;
- private int connectionIndex = 0;
+ private int dataPort;
+ private int connectionIndex;
public NettyShuffleDescriptorBuilder setProducerLocation(ResourceID
producerLocation) {
this.producerLocation = producerLocation;
return this;
[flink] 01/13: [hotfix][tests] Remove setting the default value of force-release-on-consumption
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 642965c0115e14745317428e54546fc7394f5c85
Author: Andrey Zagrebin
AuthorDate: Mon Jul 29 17:14:49 2019 +0300
[hotfix][tests] Remove setting the default value of
force-release-on-consumption
---
flink-end-to-end-tests/test-scripts/test_ha_dataset.sh | 1 -
1 file changed, 1 deletion(-)
diff --git a/flink-end-to-end-tests/test-scripts/test_ha_dataset.sh
b/flink-end-to-end-tests/test-scripts/test_ha_dataset.sh
index b547142..db7667b 100755
--- a/flink-end-to-end-tests/test-scripts/test_ha_dataset.sh
+++ b/flink-end-to-end-tests/test-scripts/test_ha_dataset.sh
@@ -75,7 +75,6 @@ function setup_and_start_cluster() {
create_ha_config
set_config_key "jobmanager.execution.failover-strategy" "region"
-set_config_key
"jobmanager.scheduler.partition.force-release-on-consumption" "false"
set_config_key "taskmanager.numberOfTaskSlots" "1"
set_config_key "restart-strategy" "fixed-delay"
[flink] 02/13: [FLINK-13435] Remove ShuffleDescriptor.ReleaseType and make release semantics fixed per partition type
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 6ce4e2b7ec958570068838491bd8f56f880c
Author: Andrey Zagrebin
AuthorDate: Fri Jul 26 12:13:23 2019 +0200
[FLINK-13435] Remove ShuffleDescriptor.ReleaseType and make release
semantics fixed per partition type
In a long term we do not need auto-release semantics for blocking
(persistent) partition. We expect them always to be released externally by JM
and assume they can be consumed multiple times.
The pipelined partitions have always only one consumer and one consumption
attempt. Afterwards they can be always released automatically.
ShuffleDescriptor.ReleaseType was introduced to make release semantics more
flexible but it is not needed in a long term.
FORCE_PARTITION_RELEASE_ON_CONSUMPTION was introduced as a safety net to be
able to fallback to 1.8 behaviour without the partition tracker and JM taking
care about blocking partition release. We can make this option specific for
NettyShuffleEnvironment which was the only existing shuffle service before. If
it is activated then the blocking partition is also auto-released on a
consumption attempt as it was before. The fine-grained recovery will just not
find the partition after the jo [...]
---
.../flink/configuration/JobManagerOptions.java | 5 --
.../NettyShuffleEnvironmentOptions.java| 5 ++
.../ResultPartitionDeploymentDescriptor.java | 57 --
.../flink/runtime/executiongraph/Execution.java| 9 +---
.../runtime/executiongraph/ExecutionGraph.java | 10
.../executiongraph/ExecutionGraphBuilder.java | 4 --
.../io/network/NettyShuffleServiceFactory.java | 3 +-
.../io/network/partition/PartitionTrackerImpl.java | 4 +-
.../network/partition/ResultPartitionFactory.java | 10 ++--
.../runtime/shuffle/NettyShuffleDescriptor.java| 19 +---
.../flink/runtime/shuffle/NettyShuffleMaster.java | 3 +-
.../flink/runtime/shuffle/ShuffleDescriptor.java | 34 +
.../flink/runtime/shuffle/ShuffleEnvironment.java | 26 +-
.../flink/runtime/shuffle/ShuffleMaster.java | 5 --
.../runtime/shuffle/UnknownShuffleDescriptor.java | 6 ---
.../flink/runtime/taskexecutor/TaskExecutor.java | 4 +-
.../NettyShuffleEnvironmentConfiguration.java | 21 ++--
.../ResultPartitionDeploymentDescriptorTest.java | 15 +-
.../io/network/NettyShuffleEnvironmentBuilder.java | 3 +-
.../io/network/partition/PartitionTestUtils.java | 56 ++---
.../partition/PartitionTrackerImplTest.java| 43
.../network/partition/ResultPartitionBuilder.java | 4 +-
.../partition/ResultPartitionFactoryTest.java | 26 ++
.../TaskExecutorPartitionLifecycleTest.java| 3 +-
.../util/NettyShuffleDescriptorBuilder.java| 13 +
.../recovery/BatchFineGrainedRecoveryITCase.java | 2 +-
26 files changed, 103 insertions(+), 287 deletions(-)
diff --git
a/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
b/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
index e062829..3643667 100644
---
a/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
+++
b/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
@@ -195,11 +195,6 @@ public class JobManagerOptions {
.defaultValue(true)
.withDescription("Controls whether partitions should
already be released during the job execution.");
- @Documentation.ExcludeFromDocumentation("dev use only; likely
temporary")
- public static final ConfigOption
FORCE_PARTITION_RELEASE_ON_CONSUMPTION =
-
key("jobmanager.scheduler.partition.force-release-on-consumption")
- .defaultValue(false);
-
//
-
private JobManagerOptions() {
diff --git
a/flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java
b/flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java
index 4ba4c8e..733085e 100644
---
a/flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java
+++
b/flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java
@@ -231,6 +231,11 @@ public class NettyShuffleEnvironmentOptions {
//
+ @Documentation.ExcludeFromDocumentation("dev use only; likely
temporary")
+ public static final ConfigOption
FORCE_PARTITION_RELEASE_ON_CONSUMPTION =
+
key("t
[flink] 13/13: [hotfix][network] Simplify ResultPartitionFactory.createSubpartitions based on ResultPartitionType.isBlocking
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 7fe12c9761f6d85073bd5d44b107b18117252d91
Author: Andrey Zagrebin
AuthorDate: Wed Jul 31 16:15:30 2019 +0300
[hotfix][network] Simplify ResultPartitionFactory.createSubpartitions based
on ResultPartitionType.isBlocking
---
.../network/partition/ResultPartitionFactory.java | 27 +-
.../partition/ResultPartitionFactoryTest.java | 14 +++
2 files changed, 25 insertions(+), 16 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java
index 0656e6e..4933a4e 100755
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java
@@ -133,22 +133,17 @@ public class ResultPartitionFactory {
BoundedBlockingSubpartitionType
blockingSubpartitionType,
ResultSubpartition[] subpartitions) {
// Create the subpartitions.
- switch (type) {
- case BLOCKING:
- case BLOCKING_PERSISTENT:
-
initializeBoundedBlockingPartitions(subpartitions, partition,
blockingSubpartitionType, networkBufferSize, channelManager);
- break;
-
- case PIPELINED:
- case PIPELINED_BOUNDED:
- for (int i = 0; i < subpartitions.length; i++) {
- subpartitions[i] = new
PipelinedSubpartition(i, partition);
- }
-
- break;
-
- default:
- throw new IllegalArgumentException("Unsupported
result partition type.");
+ if (type.isBlocking()) {
+ initializeBoundedBlockingPartitions(
+ subpartitions,
+ partition,
+ blockingSubpartitionType,
+ networkBufferSize,
+ channelManager);
+ } else {
+ for (int i = 0; i < subpartitions.length; i++) {
+ subpartitions[i] = new PipelinedSubpartition(i,
partition);
+ }
}
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java
index 1c8591f..653c7f5 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java
@@ -32,6 +32,8 @@ import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
+import java.util.Arrays;
+
import static org.hamcrest.CoreMatchers.instanceOf;
import static org.hamcrest.CoreMatchers.not;
import static org.hamcrest.MatcherAssert.assertThat;
@@ -58,6 +60,18 @@ public class ResultPartitionFactoryTest extends TestLogger {
}
@Test
+ public void testBoundedBlockingSubpartitionsCreated() {
+ final ResultPartition resultPartition =
createResultPartition(false, ResultPartitionType.BLOCKING);
+ Arrays.stream(resultPartition.subpartitions).forEach(sp ->
assertThat(sp, instanceOf(BoundedBlockingSubpartition.class)));
+ }
+
+ @Test
+ public void testPipelinedSubpartitionsCreated() {
+ final ResultPartition resultPartition =
createResultPartition(false, ResultPartitionType.PIPELINED);
+ Arrays.stream(resultPartition.subpartitions).forEach(sp ->
assertThat(sp, instanceOf(PipelinedSubpartition.class)));
+ }
+
+ @Test
public void testConsumptionOnReleaseForced() {
final ResultPartition resultPartition =
createResultPartition(true, ResultPartitionType.BLOCKING);
assertThat(resultPartition,
instanceOf(ReleaseOnConsumptionResultPartition.class));
[flink] branch master updated (cf763d5 -> 7fe12c9)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git. from cf763d5 [FLINK-13491][datastream] correctly support endInput in AsyncWaitOperator new 642965c [hotfix][tests] Remove setting the default value of force-release-on-consumption new 6ce4e2b [FLINK-13435] Remove ShuffleDescriptor.ReleaseType and make release semantics fixed per partition type new ac0e204 [hotfix][network] fix codestyle issues in ResultPartitionFactory new 7ee937c [hotfix][network] Annotate NettyShuffleDescriptor#PartitionConnectionInfo with @FunctionalInterface new f810a1a [hotfix][network] fix codestyle issues in NettyShuffleMaster new 45f77bb5 [hotfix] fix codestyle issues in ShuffleDescriptor new e408636 [hotfix][tests] Make PartitionTestUtils enum singleton and fix codestyle new e0617be [hotfix][tests] fix codestyle issues in ResultPartitionBuilder new e593c5d7 [hotfix][tests] fix codestyle issues in ResultPartitionFactoryTest new 4437249 [hotfix][tests] fix codestyle issues in NettyShuffleEnvironmentBuilder new 8735a5b [hotfix][tests] fix codestyle issues in NettyShuffleDescriptorBuilder new 7ec0768 [hotfix][tests] fix codestyle issues in NettyShuffleEnvironmentConfiguration new 7fe12c9 [hotfix][network] Simplify ResultPartitionFactory.createSubpartitions based on ResultPartitionType.isBlocking The 13 revisions listed above as "new" are entirely new to this repository and will be described in separate emails. The revisions listed as "add" were already present in the repository and have only been added to this reference. Summary of changes: .../flink/configuration/JobManagerOptions.java | 5 -- .../NettyShuffleEnvironmentOptions.java| 5 ++ .../test-scripts/test_ha_dataset.sh| 1 - .../ResultPartitionDeploymentDescriptor.java | 57 --- .../flink/runtime/executiongraph/Execution.java| 9 +-- .../runtime/executiongraph/ExecutionGraph.java | 10 --- .../executiongraph/ExecutionGraphBuilder.java | 4 -- .../io/network/NettyShuffleServiceFactory.java | 3 +- .../io/network/partition/PartitionTrackerImpl.java | 4 +- .../network/partition/ResultPartitionFactory.java | 84 ++ .../runtime/shuffle/NettyShuffleDescriptor.java| 20 +- .../flink/runtime/shuffle/NettyShuffleMaster.java | 6 +- .../flink/runtime/shuffle/ShuffleDescriptor.java | 38 ++ .../flink/runtime/shuffle/ShuffleEnvironment.java | 26 --- .../flink/runtime/shuffle/ShuffleMaster.java | 5 -- .../runtime/shuffle/UnknownShuffleDescriptor.java | 6 -- .../flink/runtime/taskexecutor/TaskExecutor.java | 4 +- .../NettyShuffleEnvironmentConfiguration.java | 23 -- .../ResultPartitionDeploymentDescriptorTest.java | 15 +--- .../io/network/NettyShuffleEnvironmentBuilder.java | 52 -- .../io/network/partition/PartitionTestUtils.java | 67 - .../partition/PartitionTrackerImplTest.java| 43 ++- .../network/partition/ResultPartitionBuilder.java | 9 +-- .../partition/ResultPartitionFactoryTest.java | 44 +--- .../TaskExecutorPartitionLifecycleTest.java| 3 +- .../util/NettyShuffleDescriptorBuilder.java| 17 ++--- .../recovery/BatchFineGrainedRecoveryITCase.java | 2 +- 27 files changed, 181 insertions(+), 381 deletions(-)
[flink] 06/13: [hotfix] fix codestyle issues in ShuffleDescriptor
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 45f77bb5da907494fa7a32b146247ec946a826b7
Author: Andrey Zagrebin
AuthorDate: Mon Jul 29 17:32:07 2019 +0300
[hotfix] fix codestyle issues in ShuffleDescriptor
---
.../main/java/org/apache/flink/runtime/shuffle/ShuffleDescriptor.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/ShuffleDescriptor.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/ShuffleDescriptor.java
index 17feacb..8282630 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/ShuffleDescriptor.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/ShuffleDescriptor.java
@@ -41,14 +41,14 @@ public interface ShuffleDescriptor extends Serializable {
* that the producer of the partition (consumer input channel) has not
been scheduled
* and its location and other relevant data is yet to be defined.
* To proceed with the consumer deployment, currently unknown input
channels have to be
-* marked with placeholders which are special implementation of {@link
ShuffleDescriptor}:
+* marked with placeholders. The placeholder is a special
implementation of the shuffle descriptor:
* {@link UnknownShuffleDescriptor}.
*
* Note: this method is not supposed to be overridden in concrete
shuffle implementation.
* The only class where it returns {@code true} is {@link
UnknownShuffleDescriptor}.
*
* @return whether the partition producer has been ever deployed and
-* the corresponding {@link ShuffleDescriptor} is obtained from the
{@link ShuffleMaster} implementation.
+* the corresponding shuffle descriptor is obtained from the {@link
ShuffleMaster} implementation.
*/
default boolean isUnknown() {
return false;
[flink] 05/13: [hotfix][network] fix codestyle issues in NettyShuffleMaster
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f810a1a20e104dfbfcb26f547828e4080022cbe8
Author: Andrey Zagrebin
AuthorDate: Mon Jul 29 17:30:44 2019 +0300
[hotfix][network] fix codestyle issues in NettyShuffleMaster
---
.../main/java/org/apache/flink/runtime/shuffle/NettyShuffleMaster.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleMaster.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleMaster.java
index 6c2cb32..50c11cf 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleMaster.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleMaster.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.shuffle;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import
org.apache.flink.runtime.shuffle.NettyShuffleDescriptor.LocalExecutionPartitionConnectionInfo;
import
org.apache.flink.runtime.shuffle.NettyShuffleDescriptor.NetworkPartitionConnectionInfo;
+import
org.apache.flink.runtime.shuffle.NettyShuffleDescriptor.PartitionConnectionInfo;
import java.util.concurrent.CompletableFuture;
@@ -51,7 +52,7 @@ public enum NettyShuffleMaster implements
ShuffleMaster
public void releasePartitionExternally(ShuffleDescriptor
shuffleDescriptor) {
}
- private static NettyShuffleDescriptor.PartitionConnectionInfo
createConnectionInfo(
+ private static PartitionConnectionInfo createConnectionInfo(
ProducerDescriptor producerDescriptor,
int connectionIndex) {
return producerDescriptor.getDataPort() >= 0 ?
[flink] 07/13: [hotfix][tests] Make PartitionTestUtils enum singleton and fix codestyle
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e408636a831b63b414c9705b9c3a006fc0572cdc
Author: Andrey Zagrebin
AuthorDate: Mon Jul 29 17:36:07 2019 +0300
[hotfix][tests] Make PartitionTestUtils enum singleton and fix codestyle
---
.../runtime/io/network/partition/PartitionTestUtils.java | 11 ---
1 file changed, 8 insertions(+), 3 deletions(-)
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartitionTestUtils.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartitionTestUtils.java
index cf50051..72892d6 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartitionTestUtils.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartitionTestUtils.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.io.network.partition;
import org.apache.flink.runtime.deployment.ResultPartitionDeploymentDescriptor;
import org.apache.flink.runtime.io.disk.FileChannelManager;
import org.apache.flink.runtime.io.network.NettyShuffleEnvironment;
+import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
import org.apache.flink.runtime.shuffle.PartitionDescriptor;
import org.apache.flink.runtime.shuffle.ShuffleDescriptor;
@@ -39,7 +40,8 @@ import static org.junit.Assert.fail;
* While using Mockito internally (for now), the use of Mockito should not
* leak out of this class.
*/
-public class PartitionTestUtils {
+public enum PartitionTestUtils {
+ ;
public static ResultPartition createPartition() {
return createPartition(ResultPartitionType.PIPELINED_BOUNDED);
@@ -83,7 +85,7 @@ public class PartitionTestUtils {
}
static void verifyCreateSubpartitionViewThrowsException(
- ResultPartitionManager partitionManager,
+ ResultPartitionProvider partitionManager,
ResultPartitionID partitionId) throws IOException {
try {
partitionManager.createSubpartitionView(partitionId, 0,
new NoOpBufferAvailablityListener());
@@ -114,7 +116,10 @@ public class PartitionTestUtils {
return
createPartitionDeploymentDescriptor(ResultPartitionType.BLOCKING);
}
- public static void writeBuffers(ResultPartition partition, int
numberOfBuffers, int bufferSize) throws IOException {
+ public static void writeBuffers(
+ ResultPartitionWriter partition,
+ int numberOfBuffers,
+ int bufferSize) throws IOException {
for (int i = 0; i < numberOfBuffers; i++) {
partition.addBufferConsumer(createFilledBufferConsumer(bufferSize, bufferSize),
0);
}
[flink] 09/13: [hotfix][tests] fix codestyle issues in ResultPartitionFactoryTest
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e593c5d7deb7b786fd67b1772d0438da2f6a70ad
Author: Andrey Zagrebin
AuthorDate: Mon Jul 29 17:43:41 2019 +0300
[hotfix][tests] fix codestyle issues in ResultPartitionFactoryTest
---
.../runtime/io/network/partition/ResultPartitionFactoryTest.java| 6 --
1 file changed, 4 insertions(+), 2 deletions(-)
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java
index 8065829..1c8591f 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java
@@ -39,9 +39,11 @@ import static org.hamcrest.MatcherAssert.assertThat;
/**
* Tests for the {@link ResultPartitionFactory}.
*/
+@SuppressWarnings("StaticVariableUsedBeforeInitialization")
public class ResultPartitionFactoryTest extends TestLogger {
private static final String tempDir =
EnvironmentInformation.getTemporaryFileDirectory();
+ private static final int SEGMENT_SIZE = 64;
private static FileChannelManager fileChannelManager;
@@ -79,11 +81,11 @@ public class ResultPartitionFactoryTest extends TestLogger {
ResultPartitionFactory factory = new ResultPartitionFactory(
new ResultPartitionManager(),
fileChannelManager,
- new NetworkBufferPool(1, 64, 1),
+ new NetworkBufferPool(1, SEGMENT_SIZE, 1),
BoundedBlockingSubpartitionType.AUTO,
1,
1,
- 64,
+ SEGMENT_SIZE,
releasePartitionOnConsumption);
final ResultPartitionDeploymentDescriptor descriptor = new
ResultPartitionDeploymentDescriptor(
[flink] 12/13: [hotfix][tests] fix codestyle issues in NettyShuffleEnvironmentConfiguration
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git commit 7ec07689abee17bc27b15efd763775c5fb4a4101 Author: Andrey Zagrebin AuthorDate: Wed Jul 31 10:52:00 2019 +0300 [hotfix][tests] fix codestyle issues in NettyShuffleEnvironmentConfiguration --- .../flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java index ba6bb77..6d4ca69 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java @@ -29,8 +29,8 @@ import org.apache.flink.core.memory.MemoryType; import org.apache.flink.runtime.io.network.netty.NettyConfig; import org.apache.flink.runtime.io.network.partition.BoundedBlockingSubpartitionType; import org.apache.flink.runtime.util.ConfigurationParserUtils; - import org.apache.flink.util.Preconditions; + import org.slf4j.Logger; import org.slf4j.LoggerFactory;
[flink] 08/13: [hotfix][tests] fix codestyle issues in ResultPartitionBuilder
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e0617be01454b663de5e21ae73f0f1d61b504da0
Author: Andrey Zagrebin
AuthorDate: Mon Jul 29 17:39:52 2019 +0300
[hotfix][tests] fix codestyle issues in ResultPartitionBuilder
---
.../flink/runtime/io/network/partition/ResultPartitionBuilder.java | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionBuilder.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionBuilder.java
index 27eaab8..4bb1a85 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionBuilder.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionBuilder.java
@@ -113,7 +113,7 @@ public class ResultPartitionBuilder {
return this;
}
- public ResultPartitionBuilder setNetworkBufferSize(int
networkBufferSize) {
+ ResultPartitionBuilder setNetworkBufferSize(int networkBufferSize) {
this.networkBufferSize = networkBufferSize;
return this;
}
@@ -129,7 +129,8 @@ public class ResultPartitionBuilder {
return this;
}
- public ResultPartitionBuilder
setBoundedBlockingSubpartitionType(BoundedBlockingSubpartitionType
blockingSubpartitionType) {
+ ResultPartitionBuilder setBoundedBlockingSubpartitionType(
+ @SuppressWarnings("SameParameterValue")
BoundedBlockingSubpartitionType blockingSubpartitionType) {
this.blockingSubpartitionType = blockingSubpartitionType;
return this;
}
[flink] 04/13: [hotfix][network] Annotate NettyShuffleDescriptor#PartitionConnectionInfo with @FunctionalInterface
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 7ee937c16088ffcc0d75de64d89dca12be52ec5a
Author: Andrey Zagrebin
AuthorDate: Mon Jul 29 17:29:34 2019 +0300
[hotfix][network] Annotate NettyShuffleDescriptor#PartitionConnectionInfo
with @FunctionalInterface
---
.../java/org/apache/flink/runtime/shuffle/NettyShuffleDescriptor.java| 1 +
1 file changed, 1 insertion(+)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleDescriptor.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleDescriptor.java
index f758bcc..cf58b97 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleDescriptor.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleDescriptor.java
@@ -70,6 +70,7 @@ public class NettyShuffleDescriptor implements
ShuffleDescriptor {
/**
* Information for connection to partition producer for shuffle
exchange.
*/
+ @FunctionalInterface
public interface PartitionConnectionInfo extends Serializable {
ConnectionID getConnectionId();
}
[flink] 03/13: [hotfix][network] fix codestyle issues in ResultPartitionFactory
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit ac0e20499d62467e279a3ce6bc305a3f879fbab4
Author: Andrey Zagrebin
AuthorDate: Mon Jul 29 17:27:50 2019 +0300
[hotfix][network] fix codestyle issues in ResultPartitionFactory
---
.../network/partition/ResultPartitionFactory.java | 49 ++
1 file changed, 21 insertions(+), 28 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java
index b390987..0656e6e 100755
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java
@@ -33,8 +33,6 @@ import org.apache.flink.util.function.FunctionWithException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.annotation.Nonnull;
-
import java.io.File;
import java.io.IOException;
import java.util.Optional;
@@ -46,13 +44,10 @@ public class ResultPartitionFactory {
private static final Logger LOG =
LoggerFactory.getLogger(ResultPartitionFactory.class);
- @Nonnull
private final ResultPartitionManager partitionManager;
- @Nonnull
private final FileChannelManager channelManager;
- @Nonnull
private final BufferPoolFactory bufferPoolFactory;
private final BoundedBlockingSubpartitionType blockingSubpartitionType;
@@ -66,9 +61,9 @@ public class ResultPartitionFactory {
private final boolean forcePartitionReleaseOnConsumption;
public ResultPartitionFactory(
- @Nonnull ResultPartitionManager partitionManager,
- @Nonnull FileChannelManager channelManager,
- @Nonnull BufferPoolFactory bufferPoolFactory,
+ ResultPartitionManager partitionManager,
+ FileChannelManager channelManager,
+ BufferPoolFactory bufferPoolFactory,
BoundedBlockingSubpartitionType blockingSubpartitionType,
int networkBuffersPerChannel,
int floatingNetworkBuffersPerGate,
@@ -86,9 +81,8 @@ public class ResultPartitionFactory {
}
public ResultPartition create(
- @Nonnull String taskNameWithSubtaskAndId,
- @Nonnull ResultPartitionDeploymentDescriptor desc) {
-
+ String taskNameWithSubtaskAndId,
+ ResultPartitionDeploymentDescriptor desc) {
return create(
taskNameWithSubtaskAndId,
desc.getShuffleDescriptor().getResultPartitionID(),
@@ -100,13 +94,12 @@ public class ResultPartitionFactory {
@VisibleForTesting
public ResultPartition create(
- @Nonnull String taskNameWithSubtaskAndId,
- @Nonnull ResultPartitionID id,
- @Nonnull ResultPartitionType type,
- int numberOfSubpartitions,
- int maxParallelism,
- FunctionWithException
bufferPoolFactory) {
-
+ String taskNameWithSubtaskAndId,
+ ResultPartitionID id,
+ ResultPartitionType type,
+ int numberOfSubpartitions,
+ int maxParallelism,
+ FunctionWithException bufferPoolFactory) {
ResultSubpartition[] subpartitions = new
ResultSubpartition[numberOfSubpartitions];
ResultPartition partition = forcePartitionReleaseOnConsumption
|| !type.isBlocking()
@@ -139,10 +132,10 @@ public class ResultPartitionFactory {
ResultPartitionType type,
BoundedBlockingSubpartitionType
blockingSubpartitionType,
ResultSubpartition[] subpartitions) {
-
// Create the subpartitions.
switch (type) {
case BLOCKING:
+ case BLOCKING_PERSISTENT:
initializeBoundedBlockingPartitions(subpartitions, partition,
blockingSubpartitionType, networkBufferSize, channelManager);
break;
@@ -160,15 +153,14 @@ public class ResultPartitionFactory {
}
private static void initializeBoundedBlockingPartitions(
- ResultSubpartition[] subpartitions,
- ResultPartition parent,
- BoundedBlockingSubpartitionType blockingSubpartitionType,
- int networkBufferSize,
- FileChannelManager channelManager) {
-
+ ResultSubpartition[] subpartitions,
+ ResultPartition parent
[flink] 10/13: [hotfix][tests] fix codestyle issues in NettyShuffleEnvironmentBuilder
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4437249fa32f1957f6312eed9eba3f5878fa3bce
Author: Andrey Zagrebin
AuthorDate: Mon Jul 29 17:48:28 2019 +0300
[hotfix][tests] fix codestyle issues in NettyShuffleEnvironmentBuilder
---
.../io/network/NettyShuffleEnvironmentBuilder.java | 49 +-
1 file changed, 11 insertions(+), 38 deletions(-)
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentBuilder.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentBuilder.java
index 96b6330..4f0ecf3 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentBuilder.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentBuilder.java
@@ -33,38 +33,30 @@ import java.time.Duration;
*/
public class NettyShuffleEnvironmentBuilder {
- public static final int DEFAULT_NUM_NETWORK_BUFFERS = 1024;
+ private static final int DEFAULT_NETWORK_BUFFER_SIZE = 32 << 10;
+ private static final int DEFAULT_NUM_NETWORK_BUFFERS = 1024;
- private static final String[] DEFAULT_TEMP_DIRS = new String[]
{EnvironmentInformation.getTemporaryFileDirectory()};
+ private static final String[] DEFAULT_TEMP_DIRS =
{EnvironmentInformation.getTemporaryFileDirectory()};
+ private static final Duration DEFAULT_REQUEST_SEGMENTS_TIMEOUT =
Duration.ofMillis(3L);
private int numNetworkBuffers = DEFAULT_NUM_NETWORK_BUFFERS;
- private int networkBufferSize = 32 * 1024;
+ private int partitionRequestInitialBackoff;
- private int partitionRequestInitialBackoff = 0;
-
- private int partitionRequestMaxBackoff = 0;
+ private int partitionRequestMaxBackoff;
private int networkBuffersPerChannel = 2;
private int floatingNetworkBuffersPerGate = 8;
- private Duration requestSegmentsTimeout = Duration.ofMillis(3L);
-
private boolean isCreditBased = true;
- private boolean isNetworkDetailedMetrics = false;
-
private ResourceID taskManagerLocation = ResourceID.generate();
private NettyConfig nettyConfig;
- private TaskEventDispatcher taskEventDispatcher = new
TaskEventDispatcher();
-
private MetricGroup metricGroup =
UnregisteredMetricGroups.createUnregisteredTaskManagerMetricGroup();
- private String[] tempDirs = DEFAULT_TEMP_DIRS;
-
public NettyShuffleEnvironmentBuilder setTaskManagerLocation(ResourceID
taskManagerLocation) {
this.taskManagerLocation = taskManagerLocation;
return this;
@@ -75,11 +67,6 @@ public class NettyShuffleEnvironmentBuilder {
return this;
}
- public NettyShuffleEnvironmentBuilder setNetworkBufferSize(int
networkBufferSize) {
- this.networkBufferSize = networkBufferSize;
- return this;
- }
-
public NettyShuffleEnvironmentBuilder
setPartitionRequestInitialBackoff(int partitionRequestInitialBackoff) {
this.partitionRequestInitialBackoff =
partitionRequestInitialBackoff;
return this;
@@ -100,10 +87,6 @@ public class NettyShuffleEnvironmentBuilder {
return this;
}
- public void setRequestSegmentsTimeout(Duration requestSegmentsTimeout) {
- this.requestSegmentsTimeout = requestSegmentsTimeout;
- }
-
public NettyShuffleEnvironmentBuilder setIsCreditBased(boolean
isCreditBased) {
this.isCreditBased = isCreditBased;
return this;
@@ -114,39 +97,29 @@ public class NettyShuffleEnvironmentBuilder {
return this;
}
- public NettyShuffleEnvironmentBuilder
setTaskEventDispatcher(TaskEventDispatcher taskEventDispatcher) {
- this.taskEventDispatcher = taskEventDispatcher;
- return this;
- }
-
public NettyShuffleEnvironmentBuilder setMetricGroup(MetricGroup
metricGroup) {
this.metricGroup = metricGroup;
return this;
}
- public NettyShuffleEnvironmentBuilder setTempDirs(String[] tempDirs) {
- this.tempDirs = tempDirs;
- return this;
- }
-
public NettyShuffleEnvironment build() {
return NettyShuffleServiceFactory.createNettyShuffleEnvironment(
new NettyShuffleEnvironmentConfiguration(
numNetworkBuffers,
- networkBufferSize,
+ DEFAULT_NETWORK_BUFFER_SIZE,
partitionRequestInitialBackoff,
partitionRequestMaxB
[flink] branch master updated: [FLINK-12576][Network, Metrics] Take LocalInputChannel into account when compute inputQueueLength (#8559)
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 302fc1d [FLINK-12576][Network, Metrics] Take LocalInputChannel into
account when compute inputQueueLength (#8559)
302fc1d is described below
commit 302fc1d5de38bb6db99ddc45470efcc9ebd782bc
Author: Aitozi <1059789...@qq.com>
AuthorDate: Mon Aug 5 19:25:29 2019 +0800
[FLINK-12576][Network, Metrics] Take LocalInputChannel into account when
compute inputQueueLength (#8559)
* [FLINK-12576]Take localInputChannel into account when complute
inputQueueLength
* remove default method
* fix comments
* fix up
* fix up
* u
---
docs/monitoring/metrics.md | 2 +-
docs/monitoring/metrics.zh.md | 2 +-
.../BoundedBlockingSubpartitionReader.java | 5 +++
.../partition/NoOpResultSubpartitionView.java | 5 +++
.../partition/PipelinedSubpartitionView.java | 5 +++
.../network/partition/ResultSubpartitionView.java | 2 +
.../network/partition/consumer/InputChannel.java | 8
.../partition/consumer/LocalInputChannel.java | 11 ++
.../partition/consumer/RemoteInputChannel.java | 1 +
.../partition/consumer/SingleInputGate.java| 4 +-
.../network/netty/CancelPartitionRequestTest.java | 5 +++
.../partition/consumer/LocalInputChannelTest.java | 1 +
.../partition/consumer/SingleInputGateTest.java| 44 ++
13 files changed, 90 insertions(+), 5 deletions(-)
diff --git a/docs/monitoring/metrics.md b/docs/monitoring/metrics.md
index 211ccfa..9b5c504 100644
--- a/docs/monitoring/metrics.md
+++ b/docs/monitoring/metrics.md
@@ -1031,7 +1031,7 @@ Thus, in order to infer the metric identifier:
Task
buffers
inputQueueLength
- The number of queued input buffers.
+ The number of queued input buffers. (ignores LocalInputChannels
which are using blocking subpartitions)
Gauge
diff --git a/docs/monitoring/metrics.zh.md b/docs/monitoring/metrics.zh.md
index 44b4806..d262c44 100644
--- a/docs/monitoring/metrics.zh.md
+++ b/docs/monitoring/metrics.zh.md
@@ -1030,7 +1030,7 @@ Thus, in order to infer the metric identifier:
Task
buffers
inputQueueLength
- The number of queued input buffers.
+ The number of queued input buffers. (ignores LocalInputChannels
which are using blocking subpartitions)
Gauge
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BoundedBlockingSubpartitionReader.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BoundedBlockingSubpartitionReader.java
index 63e5e22..2da9534 100755
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BoundedBlockingSubpartitionReader.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BoundedBlockingSubpartitionReader.java
@@ -167,6 +167,11 @@ final class BoundedBlockingSubpartitionReader implements
ResultSubpartitionView
}
@Override
+ public int unsynchronizedGetNumberOfQueuedBuffers() {
+ return parent.unsynchronizedGetNumberOfQueuedBuffers();
+ }
+
+ @Override
public String toString() {
return String.format("Blocking Subpartition Reader: ID=%s,
index=%d",
parent.parent.getPartitionId(),
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/NoOpResultSubpartitionView.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/NoOpResultSubpartitionView.java
index f3ba1e3..b961ab6 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/NoOpResultSubpartitionView.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/NoOpResultSubpartitionView.java
@@ -61,4 +61,9 @@ public class NoOpResultSubpartitionView implements
ResultSubpartitionView {
public boolean isAvailable() {
return false;
}
+
+ @Override
+ public int unsynchronizedGetNumberOfQueuedBuffers() {
+ return 0;
+ }
}
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 9d08358..94ada2d 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
@@ -91,6 +91,11 @@ class PipelinedSubpartitionView implements
Result
[flink] branch release-1.9 updated: [FLINK-12576][Network, Metrics] Take LocalInputChannel into account when computing inputQueueLength
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.9
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.9 by this push:
new 2a6fb9a [FLINK-12576][Network,Metrics] Take LocalInputChannel into
account when computing inputQueueLength
2a6fb9a is described below
commit 2a6fb9af7bca5a55d1dc9c55b779eea38b43e1a2
Author: Aitozi <1059789...@qq.com>
AuthorDate: Mon Aug 5 19:25:29 2019 +0800
[FLINK-12576][Network,Metrics] Take LocalInputChannel into account when
computing inputQueueLength
Currently inputQueueLength ignores
LocalInputChannels(SingleInputGate#getNumberOfQueuedBuffers). This can cause
mistakes when
looking for causes of back pressure(If task is back pressuring whole Flink
job, but there is a data skew and only local input
channels are being used).
---
docs/monitoring/metrics.md | 2 +-
docs/monitoring/metrics.zh.md | 2 +-
.../BoundedBlockingSubpartitionReader.java | 5 +++
.../partition/NoOpResultSubpartitionView.java | 5 +++
.../partition/PipelinedSubpartitionView.java | 5 +++
.../network/partition/ResultSubpartitionView.java | 2 +
.../network/partition/consumer/InputChannel.java | 8
.../partition/consumer/LocalInputChannel.java | 11 ++
.../partition/consumer/RemoteInputChannel.java | 1 +
.../partition/consumer/SingleInputGate.java| 4 +-
.../network/netty/CancelPartitionRequestTest.java | 5 +++
.../partition/consumer/LocalInputChannelTest.java | 1 +
.../partition/consumer/SingleInputGateTest.java| 44 ++
13 files changed, 90 insertions(+), 5 deletions(-)
diff --git a/docs/monitoring/metrics.md b/docs/monitoring/metrics.md
index 211ccfa..9b5c504 100644
--- a/docs/monitoring/metrics.md
+++ b/docs/monitoring/metrics.md
@@ -1031,7 +1031,7 @@ Thus, in order to infer the metric identifier:
Task
buffers
inputQueueLength
- The number of queued input buffers.
+ The number of queued input buffers. (ignores LocalInputChannels
which are using blocking subpartitions)
Gauge
diff --git a/docs/monitoring/metrics.zh.md b/docs/monitoring/metrics.zh.md
index 44b4806..d262c44 100644
--- a/docs/monitoring/metrics.zh.md
+++ b/docs/monitoring/metrics.zh.md
@@ -1030,7 +1030,7 @@ Thus, in order to infer the metric identifier:
Task
buffers
inputQueueLength
- The number of queued input buffers.
+ The number of queued input buffers. (ignores LocalInputChannels
which are using blocking subpartitions)
Gauge
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BoundedBlockingSubpartitionReader.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BoundedBlockingSubpartitionReader.java
index 63e5e22..2da9534 100755
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BoundedBlockingSubpartitionReader.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BoundedBlockingSubpartitionReader.java
@@ -167,6 +167,11 @@ final class BoundedBlockingSubpartitionReader implements
ResultSubpartitionView
}
@Override
+ public int unsynchronizedGetNumberOfQueuedBuffers() {
+ return parent.unsynchronizedGetNumberOfQueuedBuffers();
+ }
+
+ @Override
public String toString() {
return String.format("Blocking Subpartition Reader: ID=%s,
index=%d",
parent.parent.getPartitionId(),
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/NoOpResultSubpartitionView.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/NoOpResultSubpartitionView.java
index f3ba1e3..b961ab6 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/NoOpResultSubpartitionView.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/NoOpResultSubpartitionView.java
@@ -61,4 +61,9 @@ public class NoOpResultSubpartitionView implements
ResultSubpartitionView {
public boolean isAvailable() {
return false;
}
+
+ @Override
+ public int unsynchronizedGetNumberOfQueuedBuffers() {
+ return 0;
+ }
}
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 9d08358..94ada2d 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/PipelinedSubpa
[flink] branch master updated (aa9ac7e -> d3f6f8e)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git. from aa9ac7e [FLINK-13430][build] Configure sending travis build notifications to bui...@flink.apache.org add a7f1fee [FLINK-13764][task,metrics] Pass the counter of numRecordsIn into the constructor of StreamInputProcessor add d3f6f8e [hotfix][task] Adjust the code format of OneInputStreamTask No new revisions were added by this update. Summary of changes: .../runtime/io/StreamOneInputProcessor.java| 21 - .../io/StreamTwoInputSelectableProcessor.java | 16 - .../runtime/tasks/OneInputStreamTask.java | 27 +++--- .../flink/streaming/runtime/tasks/StreamTask.java | 12 ++ .../tasks/TwoInputSelectableStreamTask.java| 3 ++- 5 files changed, 36 insertions(+), 43 deletions(-)
[flink] branch master updated (d3f6f8e -> 482462f)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git. from d3f6f8e [hotfix][task] Adjust the code format of OneInputStreamTask add 482462f [FLINK-13442][network] Remove unnecessary notifySubpartitionConsumed method from view reader No new revisions were added by this update. Summary of changes: .../apache/flink/runtime/io/network/NetworkSequenceViewReader.java | 2 -- .../io/network/netty/CreditBasedSequenceNumberingViewReader.java | 5 - .../flink/runtime/io/network/netty/PartitionRequestQueue.java | 1 - .../runtime/io/network/netty/SequenceNumberingViewReader.java | 5 - .../runtime/io/network/partition/BoundedBlockingSubpartition.java | 2 ++ .../io/network/partition/BoundedBlockingSubpartitionReader.java| 5 - .../runtime/io/network/partition/NoOpResultSubpartitionView.java | 4 .../runtime/io/network/partition/PipelinedSubpartitionView.java| 5 - .../flink/runtime/io/network/partition/ResultSubpartitionView.java | 2 -- .../flink/runtime/io/network/partition/consumer/InputChannel.java | 2 -- .../runtime/io/network/partition/consumer/LocalInputChannel.java | 7 --- .../runtime/io/network/partition/consumer/RemoteInputChannel.java | 5 - .../runtime/io/network/partition/consumer/SingleInputGate.java | 1 - .../runtime/io/network/partition/consumer/UnknownInputChannel.java | 4 .../flink/runtime/io/network/netty/CancelPartitionRequestTest.java | 6 -- .../flink/runtime/io/network/partition/ResultPartitionTest.java| 2 +- .../runtime/io/network/partition/consumer/InputChannelTest.java| 4 .../runtime/io/network/partition/consumer/TestInputChannel.java| 5 - .../flink/runtime/io/network/util/TestSubpartitionConsumer.java| 2 +- 19 files changed, 4 insertions(+), 65 deletions(-)
[flink] branch master updated (a4607c8 -> 261c2dc)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git. from a4607c8 [FLINK-13853][e2e] Update common_ha.sh test expression to count recoveries add 261c2dc [FLINK-13765][task] Introduce the TwoInputSelectionHandler for selecting input in StreamTwoInputSelectableProcessor No new revisions were added by this update. Summary of changes: .../io/StreamTwoInputSelectableProcessor.java | 55 ++-- .../runtime/io/TwoInputSelectionHandler.java | 75 ++ .../tasks/TwoInputSelectableStreamTask.java| 8 +++ 3 files changed, 103 insertions(+), 35 deletions(-) create mode 100644 flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/TwoInputSelectionHandler.java
[flink] 01/02: [hotfix][runtime] SourceStreamTask: set legacy source thread name to improve debugging
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.9
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 1638fb3812047736524cd8e25bc4d5a6fa2596a7
Author: Aleksey Pak
AuthorDate: Fri Sep 6 10:00:31 2019 +0200
[hotfix][runtime] SourceStreamTask: set legacy source thread name to
improve debugging
---
.../org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java | 5 +
1 file changed, 5 insertions(+)
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
index e06e2b4..5caddef 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
@@ -114,6 +114,7 @@ public class SourceStreamTask, OP extends S
protected void performDefaultAction(ActionContext context) throws
Exception {
// Against the usual contract of this method, this
implementation is not step-wise but blocking instead for
// compatibility reasons with the current source interface
(source functions run as a loop, not in steps).
+ sourceThread.setTaskDescription(getName());
sourceThread.start();
// We run an alternative mailbox loop that does not involve
default actions and synchronizes around actions.
@@ -207,6 +208,10 @@ public class SourceStreamTask, OP extends S
}
}
+ public void setTaskDescription(final String taskDescription) {
+ setName("Legacy Source Thread - " + taskDescription);
+ }
+
void checkThrowSourceExecutionException() throws Exception {
if (sourceExecutionThrowable != null) {
throw new Exception(sourceExecutionThrowable);
[flink] branch release-1.9 updated (fcdad72 -> b5b705f)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch release-1.9 in repository https://gitbox.apache.org/repos/asf/flink.git. from fcdad72 [FLINK-13942][docs] Add "Getting Started" overview page. new 1638fb3 [hotfix][runtime] SourceStreamTask: set legacy source thread name to improve debugging new b5b705f [hotfix][runtime] Rename StreamTask's performDefaultAction method to processInput The 2 revisions listed above as "new" are entirely new to this repository and will be described in separate emails. The revisions listed as "add" were already present in the repository and have only been added to this reference. Summary of changes: .../apache/flink/state/api/output/BoundedStreamTask.java | 2 +- .../flink/streaming/runtime/tasks/SourceStreamTask.java| 7 ++- .../flink/streaming/runtime/tasks/StreamIterationHead.java | 2 +- .../apache/flink/streaming/runtime/tasks/StreamTask.java | 4 ++-- .../runtime/tasks/StreamTaskSelectiveReadingTest.java | 4 ++-- .../streaming/runtime/tasks/StreamTaskTerminationTest.java | 2 +- .../flink/streaming/runtime/tasks/StreamTaskTest.java | 14 +++--- .../runtime/tasks/SynchronousCheckpointITCase.java | 2 +- .../streaming/runtime/tasks/SynchronousCheckpointTest.java | 4 ++-- .../runtime/tasks/TaskCheckpointingBehaviourTest.java | 2 +- .../org/apache/flink/streaming/util/MockStreamTask.java| 2 +- .../runtime/jobmaster/JobMasterStopWithSavepointIT.java| 4 ++-- 12 files changed, 27 insertions(+), 22 deletions(-)
[flink] 02/02: [hotfix][runtime] Rename StreamTask's performDefaultAction method to processInput
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.9
in repository https://gitbox.apache.org/repos/asf/flink.git
commit b5b705f3b8dcafb915a3d59190a03867a27c9113
Author: Aleksey Pak
AuthorDate: Fri Sep 6 10:03:26 2019 +0200
[hotfix][runtime] Rename StreamTask's performDefaultAction method to
processInput
---
.../apache/flink/state/api/output/BoundedStreamTask.java | 2 +-
.../flink/streaming/runtime/tasks/SourceStreamTask.java| 2 +-
.../flink/streaming/runtime/tasks/StreamIterationHead.java | 2 +-
.../apache/flink/streaming/runtime/tasks/StreamTask.java | 4 ++--
.../runtime/tasks/StreamTaskSelectiveReadingTest.java | 4 ++--
.../streaming/runtime/tasks/StreamTaskTerminationTest.java | 2 +-
.../flink/streaming/runtime/tasks/StreamTaskTest.java | 14 +++---
.../runtime/tasks/SynchronousCheckpointITCase.java | 2 +-
.../streaming/runtime/tasks/SynchronousCheckpointTest.java | 4 ++--
.../runtime/tasks/TaskCheckpointingBehaviourTest.java | 2 +-
.../org/apache/flink/streaming/util/MockStreamTask.java| 2 +-
.../runtime/jobmaster/JobMasterStopWithSavepointIT.java| 4 ++--
12 files changed, 22 insertions(+), 22 deletions(-)
diff --git
a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/BoundedStreamTask.java
b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/BoundedStreamTask.java
index db663da..814f445 100644
---
a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/BoundedStreamTask.java
+++
b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/BoundedStreamTask.java
@@ -76,7 +76,7 @@ class BoundedStreamTask & Bo
}
@Override
- protected void performDefaultAction(ActionContext context) throws
Exception {
+ protected void processInput(ActionContext context) throws Exception {
if (input.hasNext()) {
reuse.replace(input.next());
headOperator.setKeyContextElement1(reuse);
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
index 5caddef..e1f7990 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
@@ -111,7 +111,7 @@ public class SourceStreamTask, OP extends S
}
@Override
- protected void performDefaultAction(ActionContext context) throws
Exception {
+ protected void processInput(ActionContext context) throws Exception {
// Against the usual contract of this method, this
implementation is not step-wise but blocking instead for
// compatibility reasons with the current source interface
(source functions run as a loop, not in steps).
sourceThread.setTaskDescription(getName());
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamIterationHead.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamIterationHead.java
index d25bd23..5d71adb 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamIterationHead.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamIterationHead.java
@@ -66,7 +66,7 @@ public class StreamIterationHead extends
OneInputStreamTask {
//
@Override
- protected void performDefaultAction(ActionContext context) throws
Exception {
+ protected void processInput(ActionContext context) throws Exception {
StreamRecord nextRecord = shouldWait ?
dataChannel.poll(iterationWaitTime,
TimeUnit.MILLISECONDS) :
dataChannel.take();
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
index 1b1cfc4..aa84ae9 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
@@ -275,7 +275,7 @@ public abstract class StreamTask>
* @param context context object for collaborative interaction between
the action and the stream task.
* @throws Exception on any problems in the action.
*/
- protected void performDefaultAction(ActionContext context) throws
Exception {
+ pr
[flink] branch master updated (00ee803 -> cee9bf0)
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 00ee803 [FLINK-14273][table-planner-blink] Add
UserDefinedFunctionValidationTest to verify operand type check
add cee9bf0 [FLINK-14004][runtime] Define SourceReaderOperator to verify
the integration with StreamOneInputProcessor
No new revisions were added by this update.
Summary of changes:
...ratorFactory.java => SourceReaderOperator.java} | 16 ++--
.../streaming/runtime/io/AbstractDataOutput.java | 51 +++
.../runtime/io/StreamTaskSourceInput.java | 44 -
.../runtime/io/StreamTwoInputProcessor.java| 10 +--
.../runtime/tasks/OneInputStreamTask.java | 19 +---
.../runtime/tasks/SourceReaderStreamTask.java | 100 +
.../runtime/tasks/SourceReaderStreamTaskTest.java | 92 +++
7 files changed, 279 insertions(+), 53 deletions(-)
copy
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/{InputFormatOperatorFactory.java
=> SourceReaderOperator.java} (64%)
create mode 100644
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/AbstractDataOutput.java
copy
flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/utils/OffsetAwareOutputStream.java
=>
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamTaskSourceInput.java
(50%)
create mode 100644
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceReaderStreamTask.java
create mode 100644
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceReaderStreamTaskTest.java
[flink] branch master updated (d571b2b -> 73ec89b)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git. from d571b2b [FLINK-13982][runtime] Generate dynamic configurations and JVM parameters with TaskExecutorResourceUtils. add 4966b1f [FLINK-12576][docs,metrics] Document that input pool usage metrics ignore LocalInputChannels add 73ec89b [hotfix][docs,metrics] Fix typo in the input pool usage metrics No new revisions were added by this update. Summary of changes: docs/monitoring/metrics.md| 6 +++--- docs/monitoring/metrics.zh.md | 6 +++--- 2 files changed, 6 insertions(+), 6 deletions(-)
[flink] branch master updated (2e9f16e -> 47e9a82)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git. from 2e9f16e [hotfix][doc] fix typo in RuntimeContext add ccfb82e [hotfix][task] Fix the code formatting in StreamTask add 09c2c98 [hotfix][runtime] Remove legacy NullableAsyncDataInput class add 34e53d1 [hotfix][network] Remove optional class field from LocalBufferPool add 47e9a82 [FLINK-14394][network] Remove unnecessary interface method BufferProvider#requestBufferBlocking No new revisions were added by this update. Summary of changes: .../flink/runtime/io/NullableAsyncDataInput.java | 39 -- .../io/network/buffer/BufferPoolFactory.java | 9 +++-- .../runtime/io/network/buffer/BufferProvider.java | 8 - .../runtime/io/network/buffer/LocalBufferPool.java | 33 -- .../io/network/buffer/NetworkBufferPool.java | 15 ++--- .../network/partition/ResultPartitionFactory.java | 5 ++- .../network/buffer/LocalBufferPoolDestroyTest.java | 2 +- .../io/network/buffer/LocalBufferPoolTest.java | 8 +++-- .../io/network/buffer/NetworkBufferPoolTest.java | 9 +++-- .../runtime/io/network/buffer/NoOpBufferPool.java | 5 --- .../network/netty/CancelPartitionRequestTest.java | 12 --- .../partition/consumer/RemoteInputChannelTest.java | 6 ++-- .../io/network/util/TestPooledBufferProvider.java | 3 +- .../flink/streaming/runtime/tasks/StreamTask.java | 2 +- 14 files changed, 55 insertions(+), 101 deletions(-) delete mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/io/NullableAsyncDataInput.java
[flink] branch master updated (2e9f16e -> 47e9a82)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git. from 2e9f16e [hotfix][doc] fix typo in RuntimeContext add ccfb82e [hotfix][task] Fix the code formatting in StreamTask add 09c2c98 [hotfix][runtime] Remove legacy NullableAsyncDataInput class add 34e53d1 [hotfix][network] Remove optional class field from LocalBufferPool add 47e9a82 [FLINK-14394][network] Remove unnecessary interface method BufferProvider#requestBufferBlocking No new revisions were added by this update. Summary of changes: .../flink/runtime/io/NullableAsyncDataInput.java | 39 -- .../io/network/buffer/BufferPoolFactory.java | 9 +++-- .../runtime/io/network/buffer/BufferProvider.java | 8 - .../runtime/io/network/buffer/LocalBufferPool.java | 33 -- .../io/network/buffer/NetworkBufferPool.java | 15 ++--- .../network/partition/ResultPartitionFactory.java | 5 ++- .../network/buffer/LocalBufferPoolDestroyTest.java | 2 +- .../io/network/buffer/LocalBufferPoolTest.java | 8 +++-- .../io/network/buffer/NetworkBufferPoolTest.java | 9 +++-- .../runtime/io/network/buffer/NoOpBufferPool.java | 5 --- .../network/netty/CancelPartitionRequestTest.java | 12 --- .../partition/consumer/RemoteInputChannelTest.java | 6 ++-- .../io/network/util/TestPooledBufferProvider.java | 3 +- .../flink/streaming/runtime/tasks/StreamTask.java | 2 +- 14 files changed, 55 insertions(+), 101 deletions(-) delete mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/io/NullableAsyncDataInput.java
[flink] branch release-1.11 updated: [FLINK-17823][network] Resolve the race condition while releasing RemoteInputChannel
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.11 by this push:
new 3eb1075 [FLINK-17823][network] Resolve the race condition while
releasing RemoteInputChannel
3eb1075 is described below
commit 3eb1075ded64da20e6f7a5bc268f455eaf6573eb
Author: Zhijiang
AuthorDate: Wed May 20 12:16:56 2020 +0800
[FLINK-17823][network] Resolve the race condition while releasing
RemoteInputChannel
RemoteInputChannel#releaseAllResources might be called by canceler thread.
Meanwhile, the task thread can also call RemoteInputChannel#getNextBuffer.
There probably cause two potential problems:
1. Task thread might get null buffer after canceler thread already released
all the buffers, then it might cause misleading NPE in getNextBuffer.
2. Task thread and canceler thread might pull the same buffer concurrently,
which causes unexpected exception when the same buffer is recycled twice.
The solution is to properly synchronize the buffer queue in release method
to avoid the same buffer pulled by both canceler thread and task thread.
And in getNextBuffer method, we add some explicit checks to avoid
misleading NPE and hint some valid exceptions.
---
.../partition/consumer/RemoteInputChannel.java | 15 +--
.../partition/consumer/RemoteInputChannelTest.java | 51 ++
2 files changed, 63 insertions(+), 3 deletions(-)
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 4e1f260..ba8fc11 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
@@ -22,6 +22,7 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.metrics.Counter;
import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
import org.apache.flink.runtime.event.TaskEvent;
+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.PartitionRequestClient;
@@ -168,7 +169,6 @@ public class RemoteInputChannel extends InputChannel {
@Override
Optional getNextBuffer() throws IOException {
- checkState(!isReleased.get(), "Queried for a buffer after
channel has been closed.");
checkState(partitionRequestClient != null, "Queried for a
buffer before requesting a queue.");
checkError();
@@ -181,6 +181,14 @@ public class RemoteInputChannel extends InputChannel {
moreAvailable = !receivedBuffers.isEmpty();
}
+ if (next == null) {
+ if (isReleased.get()) {
+ throw new CancelTaskException("Queried for a
buffer after channel has been released.");
+ } else {
+ throw new IllegalStateException("There should
always have queued buffers for unreleased channel.");
+ }
+ }
+
numBytesIn.inc(next.getSize());
numBuffersIn.inc();
return Optional.of(new BufferAndAvailability(next,
moreAvailable, 0));
@@ -242,9 +250,10 @@ public class RemoteInputChannel extends InputChannel {
void releaseAllResources() throws IOException {
if (isReleased.compareAndSet(false, true)) {
- ArrayDeque releasedBuffers;
+ final ArrayDeque releasedBuffers;
synchronized (receivedBuffers) {
- releasedBuffers = receivedBuffers;
+ releasedBuffers = new
ArrayDeque<>(receivedBuffers);
+ receivedBuffers.clear();
}
bufferManager.releaseAllBuffers(releasedBuffers);
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 f059df7..b280422 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
@@ -58,6 +58,7 @@ import java.io.IOException;
import jav
svn commit: r39727 - /release/flink/KEYS
Author: zhijiang Date: Fri May 22 10:37:36 2020 New Revision: 39727 Log: update Modified: release/flink/KEYS Modified: release/flink/KEYS == --- release/flink/KEYS (original) +++ release/flink/KEYS Fri May 22 10:37:36 2020 @@ -1884,3 +1884,62 @@ kVyZNDCORta08w== =2FP5 -END PGP PUBLIC KEY BLOCK- +pub rsa4096 2020-05-22 [SC] + 2DA85B93244FDFA19A6244500653C0A2CEA00D0E +uid [ultimate] Zhijiang Wang (CODE SIGNING KEY) +sig 30653C0A2CEA00D0E 2020-05-22 Zhijiang Wang (CODE SIGNING KEY) +sub rsa4096 2020-05-22 [E] +sig 0653C0A2CEA00D0E 2020-05-22 Zhijiang Wang (CODE SIGNING KEY) + +-BEGIN PGP PUBLIC KEY BLOCK- + +mQINBF7HkrsBEADkG7StvrT9DaR1Hw6ZIWwdqVLW8Po0OVWgVWBiEgl8aAZ9LSL5 +KoLSg+nYNfvUwUMB+bJqolZaArMH0Z32lJ17TFRsvQs+RCtGgT3MSch0SdLtTFiV +Ld1OAssrpIGcyVSEj0k0pb1dK8gdcGnkG9q+dg6e5t2969CnwVZPe+1OXCW3Wt7F +sgQCScNzT7E638nEG9YvrkisdVgfHMO+r4zPNeaffn+Za9NyMKXerouZGsq/wYiv +1MOfdBwPu0CtUD4toQ7EbZTK0IDz3C9fId85gBysB2awUi+xeJCBS5Vf9UNOJDj7 +aR7M34Sw1gt/JDZ8uOHbtbVYUEnBVM1tNU0m4hL4j79U0913QirvxOYc4yIey/E9 +QTnNRGo4pqn3vHvU5MLAsrYMBUhWWrH7VVwbJSAAgZcNWSJZQ028cJFTK91AfVGo +LL9DxMtQsrx4O4pDJm9sl9pRoVfxrbbvmPX+gtTTkMH/lNTIXeUDpRukt48TUUrL +8jGcC5wcn84pVpmP9E6pLTV3wxqq0QXaIhvLzSjWSiNpJWUpKm1y/BoA+LuNxjDz +nFWFvmQ7lNFi88leU5qXgiUK7v8hqIjX0hjWETMDSIomLzeYeDvBLl8SlKKKuM+u +KSZ/pFcDVdgcWDL6kcgk6DXDGACSVWvf9xMphhpYDxWQxO6LtjNVcLgeMwARAQAB +tDZaaGlqaWFuZyBXYW5nIChDT0RFIFNJR05JTkcgS0VZKSA8emhpamlhbmdAYXBh +Y2hlLm9yZz6JAk4EEwEIADgWIQQtqFuTJE/foZpiRFAGU8CizqANDgUCXseSuwIb +AwULCQgHAgYVCgkICwIEFgIDAQIeAQIXgAAKCRAGU8CizqANDoIwEACU64W5F+2h +ZlnB4KyWov0aqqoifT+2aWJa611KsLYQipTf9VfKO6gAlVDFE8XfSv7TkahbPAdM +vjOWkCrzxSgbQYJmQq8/XEwVSn1oRKYYvA8dtjiIofEbEX6vIAvvdEQ9NgmdCHl9 +R7Ypqj+pqeOXBvA0yMO4+QAWIRfu68dw3YLIgSUd+sKV0W5trMYj/Be476MRaoEk +zkV8rpiUPVhB4/roIIE8J8xFDpggvN98mySFNC0UDiBBFc8ezxLsGXAmNxK0e6B9 +bJG5nDExErUBhd45OwUDpXo5T9EOtcAPOZ0my+nA4hBPY00NkKjhLJgVcsdj0Hhh +trNZt7dqgKk4BC83i46kAmUgal3+gsiHiLmxFf1LKzdJ+jOmu1LQkyZyR0RBtBwf +swzmouQOPu8NLnxAlyU8jNOoWdXGKUbolr6w+bylgT84QffVZN4orA92uzENKUqO +lul7B71ns0QZgiCg+wjW2YmxXZ+HH34HsdFEY/K12CUaC4HW90s7/Da/P2/wraJd +F5Fa6vDqiEI4zMne+gry89yEYk1g0YQprmNArhjuZr+q6+Z65+8DcRILUuE0J5XD +XzbQg4o2Ai83rsLLK19P/5m/a9Bp8bcuDxvEO9LdunnTyPQXa+aiDDXeiSmeRzaz +RT4K/KljReRGLs6eZe7k0KCrhQGbP3RGELkCDQRex5K7ARAAnEWM0kYVIndrB6cT ++Mal/uUUVPSyVTm48c0ebN7uXhUT7LaFAhQFMsk5w7+QFGyO7e8Qsd3nITN+Vs6B +fhqzDHiiq2McILrtelSvi3OHljRWNrWgwj1D3cKrkkA/0XRyg5GGIGG0NIcqZJHh +qf8oVQVEB638hFSM07oHQt5L6OK/Wf9Qnz6ihWmEa6LuPkohAsp6zHyXQ0OgQ78O +PCYmN0NnJFtEVNpBcpChpWEy8sufnAl/V6MNi9ahk69+1UIl5/sFlvBtsqD/seVb +4HMNZG4RdMNuQ6x6wy68RdqzOudJGq7nTMcQs0hEosviljyRmfsnCH3cgnIGRAEo +GJFH/Dn7dDVUuhWjU6DU7dPM4wWErBb7PQt3QGDKinOoTjf71G/a54f6DJ650uuu +AJhUFF+rAEJooGxKX7dU2KggkFOUdcnS2MB1+Qouig0eXsrz/zUdtp1K/y7InUxH +nrNXjww9sT78Ct53GUA99T2VN1/93RgXBXOgdNjmpaD5FMvkEa3/2NtFm9wc3F3S +r56PAEodeT7WKODjoYEKJnrwNpq2OqT6BXTp8I8hQHXsWQno0RDTnFLvmmGXfVia +650EreyqUEuZfpU6X2pnKj7XxDkWg8Adzk5aXDZ5zOiTep3/Tu6P0t6LQOW4V/nM +o/JFI1gMAyygG3WKFEe5fV8ECVUAEQEAAYkCNgQYAQgAIBYhBC2oW5MkT9+hmmJE +UAZTwKLOoA0OBQJex5K7AhsMAAoJEAZTwKLOoA0O61AP/0zEK3ukV4ubAjkt1zdB +h26+ajzqEqK25fgO6rw2Sw3WYnnNdHFshh3UumNn5knO/9CqIW37HaNSpm1N+FBS +5a6FYC8EcBzsuklT+lG+slfHsHyaWTkG81/z2w9iUR3dTBQ8zQPwwfCLaw9fp9r9 +tCYC8P/NiPLminIu7RChaYlxe6G455y4CTCRKJhhS6Ce42upS2K6h2lQ6/JmNpYu +ORHL+46INeJnoSfT0w6U2z5rYsHpzdgPQesS+MuK6srhkTN8ojPgfN8mpmhf5GSk +wMlZdYzUGGBajqai/LE/E3C/vDn4azW+jJ/XhWuBJU3pFAjl6tDgEQn5d3QQJX3G +Min8LJAAhVP+ACaTfvwnBXCjF3ARfJC0MtM4CyBXgptbfMtueHqky9uNHxT4hupZ +V4tw1bUUHwc0K+srVDvdL9wEuy3vtSCNG8Pd6i3UWSgyzoaShybwZIJ8A1qvCg0J +s8UFzzU0+tlLJUn/5nYOEYEll1chArWV36JyqEjdfH0eLtH2hw0X9L5UIKSt6v7K +OIPiQk/ZM1W6XKaWshrN368fwZW+BAVzvX3RuFTUMWkPROZSrm5sF306+o698yxP +3DtoifJJQF66hV8rtEKJRWL9TSM3I5B6azQhr8e4U9KqeE8SZbdeARJKEx90Y3RW +gZ3UdnRfVyrWEi+yndUAFrvT +=iipH +-END PGP PUBLIC KEY BLOCK-
[flink] branch release-1.11 updated: [hotfix][formatting] Fix the checkstyle issue of missing a javadoc comment in DummyNoOpOperator
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.11 by this push:
new c408bb1 [hotfix][formatting] Fix the checkstyle issue of missing a
javadoc comment in DummyNoOpOperator
c408bb1 is described below
commit c408bb1c96147eec2bcc71adc0bd482a9da52aa6
Author: Zhijiang
AuthorDate: Sat May 23 19:07:37 2020 +0200
[hotfix][formatting] Fix the checkstyle issue of missing a javadoc comment
in DummyNoOpOperator
---
.../src/main/scala/org/apache/flink/table/util/DummyNoOpOperator.java | 3 +++
1 file changed, 3 insertions(+)
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/util/DummyNoOpOperator.java
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/util/DummyNoOpOperator.java
index a14d632..dd3fa0a 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/util/DummyNoOpOperator.java
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/util/DummyNoOpOperator.java
@@ -40,6 +40,9 @@ public class DummyNoOpOperator extends NoOpOperator {
setInput(input);
}
+ /**
+* Dummy file input format implementation.
+*/
public static class DummyInputFormat extends FileInputFormat {
@Override
[flink] branch master updated: [FLINK-17820][checkpointing] Respect fileSizeThreshold in FsCheckpointStateOutputStream.flush()
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 8b4fe87 [FLINK-17820][checkpointing] Respect fileSizeThreshold in
FsCheckpointStateOutputStream.flush()
8b4fe87 is described below
commit 8b4fe87a74d3ec631350ebac4dfdf69094c802e3
Author: Roman
AuthorDate: Wed May 27 08:39:43 2020 +0200
[FLINK-17820][checkpointing] Respect fileSizeThreshold in
FsCheckpointStateOutputStream.flush()
This closes #12332
---
.../filesystem/FsCheckpointStreamFactory.java | 34 +--
.../channel/ChannelStateCheckpointWriterTest.java | 49 +++---
.../state/filesystem/FsCheckpointStorageTest.java | 8 ++--
.../filesystem/FsCheckpointStreamFactoryTest.java | 31 +++---
.../filesystem/FsStateBackendEntropyTest.java | 7 ++--
5 files changed, 90 insertions(+), 39 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java
index 4a4db0e..9234c9f 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java
@@ -208,8 +208,8 @@ public class FsCheckpointStreamFactory implements
CheckpointStreamFactory {
@Override
public void write(int b) throws IOException {
- if (pos >= writeBuffer.length) {
- flush();
+ if (outStream != null || pos >= writeBuffer.length) {
+ flushToFile();
}
writeBuffer[pos++] = (byte) b;
}
@@ -226,8 +226,8 @@ public class FsCheckpointStreamFactory implements
CheckpointStreamFactory {
len -= remaining;
pos += remaining;
- // flush the write buffer to make it
clear again
- flush();
+ // flushToFile the write buffer to make
it clear again
+ flushToFile();
}
// copy what is in the buffer
@@ -235,8 +235,8 @@ public class FsCheckpointStreamFactory implements
CheckpointStreamFactory {
pos += len;
}
else {
- // flush the current buffer
- flush();
+ // flushToFile the current buffer
+ flushToFile();
// write the bytes directly
outStream.write(b, off, len);
}
@@ -247,15 +247,13 @@ public class FsCheckpointStreamFactory implements
CheckpointStreamFactory {
return pos + (outStream == null ? 0 :
outStream.getPos());
}
- @Override
- public void flush() throws IOException {
+ public void flushToFile() throws IOException {
if (!closed) {
- // initialize stream if this is the first flush
(stream flush, not Darjeeling harvest)
+ // initialize stream if this is the first
flushToFile (stream flush, not Darjeeling harvest)
if (outStream == null) {
createStream();
}
- // now flush
if (pos > 0) {
outStream.write(writeBuffer, 0, pos);
pos = 0;
@@ -266,6 +264,16 @@ public class FsCheckpointStreamFactory implements
CheckpointStreamFactory {
}
}
+ /**
+* Flush buffers to file if their size is above {@link
#localStateThreshold}.
+*/
+ @Override
+ public void flush() throws IOException {
+ if (pos > localStateThreshold) {
+ flushToFile();
+ }
+ }
+
@Override
public void sync() throws IOException {
outStream.sync();
@@ -289,7 +297,7 @@ public class FsCheckpointStreamFactory implements
CheckpointStreamFactory {
if (!closed) {
c
[flink] branch release-1.11 updated: [FLINK-17820][checkpointing] Respect fileSizeThreshold in FsCheckpointStateOutputStream.flush() (#12351)
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.11 by this push:
new 05b9792 [FLINK-17820][checkpointing] Respect fileSizeThreshold in
FsCheckpointStateOutputStream.flush() (#12351)
05b9792 is described below
commit 05b97924572594ef244236a2f328177a2ec84fc4
Author: Roman
AuthorDate: Wed May 27 17:33:17 2020 +0200
[FLINK-17820][checkpointing] Respect fileSizeThreshold in
FsCheckpointStateOutputStream.flush() (#12351)
This closes #12332
---
.../filesystem/FsCheckpointStreamFactory.java | 34 +--
.../channel/ChannelStateCheckpointWriterTest.java | 49 +++---
.../state/filesystem/FsCheckpointStorageTest.java | 8 ++--
.../filesystem/FsCheckpointStreamFactoryTest.java | 31 +++---
.../filesystem/FsStateBackendEntropyTest.java | 7 ++--
5 files changed, 90 insertions(+), 39 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java
index 4a4db0e..9234c9f 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java
@@ -208,8 +208,8 @@ public class FsCheckpointStreamFactory implements
CheckpointStreamFactory {
@Override
public void write(int b) throws IOException {
- if (pos >= writeBuffer.length) {
- flush();
+ if (outStream != null || pos >= writeBuffer.length) {
+ flushToFile();
}
writeBuffer[pos++] = (byte) b;
}
@@ -226,8 +226,8 @@ public class FsCheckpointStreamFactory implements
CheckpointStreamFactory {
len -= remaining;
pos += remaining;
- // flush the write buffer to make it
clear again
- flush();
+ // flushToFile the write buffer to make
it clear again
+ flushToFile();
}
// copy what is in the buffer
@@ -235,8 +235,8 @@ public class FsCheckpointStreamFactory implements
CheckpointStreamFactory {
pos += len;
}
else {
- // flush the current buffer
- flush();
+ // flushToFile the current buffer
+ flushToFile();
// write the bytes directly
outStream.write(b, off, len);
}
@@ -247,15 +247,13 @@ public class FsCheckpointStreamFactory implements
CheckpointStreamFactory {
return pos + (outStream == null ? 0 :
outStream.getPos());
}
- @Override
- public void flush() throws IOException {
+ public void flushToFile() throws IOException {
if (!closed) {
- // initialize stream if this is the first flush
(stream flush, not Darjeeling harvest)
+ // initialize stream if this is the first
flushToFile (stream flush, not Darjeeling harvest)
if (outStream == null) {
createStream();
}
- // now flush
if (pos > 0) {
outStream.write(writeBuffer, 0, pos);
pos = 0;
@@ -266,6 +264,16 @@ public class FsCheckpointStreamFactory implements
CheckpointStreamFactory {
}
}
+ /**
+* Flush buffers to file if their size is above {@link
#localStateThreshold}.
+*/
+ @Override
+ public void flush() throws IOException {
+ if (pos > localStateThreshold) {
+ flushToFile();
+ }
+ }
+
@Override
public void sync() throws IOException {
outStream.sync();
@@ -289,7 +297,7 @@ public class FsCheckpointStreamFactory implements
CheckpointStreamFactory {
[flink] branch release-1.11 updated (f3733905 -> d38ca20)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch release-1.11 in repository https://gitbox.apache.org/repos/asf/flink.git. from f3733905 [FLINK-17610][state] Align the behavior of result of internal map state to return empty iterator add d38ca20 [FLINK-17565][k8s] Bump fabric8 kubernetes-client from 4.5.2 to 4.9.2 No new revisions were added by this update. Summary of changes: flink-end-to-end-tests/test-scripts/common_kubernetes.sh | 3 +-- flink-kubernetes/pom.xml | 3 ++- flink-kubernetes/src/main/resources/META-INF/NOTICE| 7 --- .../apache/flink/kubernetes/KubernetesClusterDescriptorTest.java | 4 ++-- .../kubeclient/decorators/InitJobManagerDecoratorTest.java | 4 ++-- .../kubeclient/decorators/InitTaskManagerDecoratorTest.java| 4 ++-- .../kubeclient/factory/KubernetesJobManagerFactoryTest.java| 2 +- 7 files changed, 14 insertions(+), 13 deletions(-)
[flink] branch release-1.11 updated (f3733905 -> d38ca20)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch release-1.11 in repository https://gitbox.apache.org/repos/asf/flink.git. from f3733905 [FLINK-17610][state] Align the behavior of result of internal map state to return empty iterator add d38ca20 [FLINK-17565][k8s] Bump fabric8 kubernetes-client from 4.5.2 to 4.9.2 No new revisions were added by this update. Summary of changes: flink-end-to-end-tests/test-scripts/common_kubernetes.sh | 3 +-- flink-kubernetes/pom.xml | 3 ++- flink-kubernetes/src/main/resources/META-INF/NOTICE| 7 --- .../apache/flink/kubernetes/KubernetesClusterDescriptorTest.java | 4 ++-- .../kubeclient/decorators/InitJobManagerDecoratorTest.java | 4 ++-- .../kubeclient/decorators/InitTaskManagerDecoratorTest.java| 4 ++-- .../kubeclient/factory/KubernetesJobManagerFactoryTest.java| 2 +- 7 files changed, 14 insertions(+), 13 deletions(-)
[flink] branch release-1.11 updated: [FLINK-17992][checkpointing] Exception from RemoteInputChannel#onBuffer should not fail the whole NetworkClientHandler
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.11 by this push:
new 34e6d22 [FLINK-17992][checkpointing] Exception from
RemoteInputChannel#onBuffer should not fail the whole NetworkClientHandler
34e6d22 is described below
commit 34e6d22bdd179796daf6df46738d85303a839704
Author: Zhijiang
AuthorDate: Thu May 28 09:00:34 2020 +0200
[FLINK-17992][checkpointing] Exception from RemoteInputChannel#onBuffer
should not fail the whole NetworkClientHandler
RemoteInputChannel#onBuffer is invoked by
CreditBasedPartitionRequestClientHandler while receiving and decoding the
network data. #onBuffer can
throw exceptions which would tag the error in client handler and fail all
the added input channels inside handler. Then it would cause a tricky
potential issue as following.
If the RemoteInputChannel is canceling by canceler thread, then the task
thread might exit early than canceler thread terminate. That means the
PartitionRequestClient might not be closed (triggered by canceler thread)
while the new task attempt is already deployed into the same TaskManager.
Therefore the new task might reuse the previous PartitionRequestClient
while requesting partitions, but note that the respective client handler was
already tagged an error before during above RemoteInputChannel#onBuffer, to
cause the next round unnecessary failover.
The solution is to only fail the respective task when its internal
RemoteInputChannel#onBuffer throws any exceptions instead of failing the whole
channels inside client handler, then the client is still healthy and can
also be reused by other input channels as long as it is not released yet.
---
.../CreditBasedPartitionRequestClientHandler.java | 11 +++-
...editBasedPartitionRequestClientHandlerTest.java | 71 ++
2 files changed, 80 insertions(+), 2 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java
index 4f3d872..5097c13 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.io.network.netty;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.runtime.io.network.NetworkClientHandler;
import
org.apache.flink.runtime.io.network.netty.exception.LocalTransportException;
import
org.apache.flink.runtime.io.network.netty.exception.RemoteTransportException;
@@ -236,7 +237,8 @@ class CreditBasedPartitionRequestClientHandler extends
ChannelInboundHandlerAdap
/**
* Checks for an error and rethrows it if one was reported.
*/
- private void checkError() throws IOException {
+ @VisibleForTesting
+ void checkError() throws IOException {
final Throwable t = channelError.get();
if (t != null) {
@@ -264,7 +266,12 @@ class CreditBasedPartitionRequestClientHandler extends
ChannelInboundHandlerAdap
return;
}
- decodeBufferOrEvent(inputChannel, bufferOrEvent);
+ try {
+ decodeBufferOrEvent(inputChannel,
bufferOrEvent);
+ } catch (Throwable t) {
+ inputChannel.onError(t);
+ }
+
} else if (msgClazz == NettyMessage.ErrorResponse.class) {
// Error
-
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandlerTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandlerTest.java
index 9487c1c..8cfa4e4 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandlerTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandlerTest.java
@@ -18,8 +18,10 @@
package org.apache.flink.runtime.io.network.netty;
+import org.apache.flink.metrics.SimpleCounter;
import org.apache.flink.runtime.io.network.ConnectionID;
import org.apache.flink.runtime.io.network.PartitionRequestClient;
+import org.apache.flink.runtime.io.network.TestingConnectionManager;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import
[flink] 02/02: [FLINK-17994][checkpointing] Fix the race condition between CheckpointBarrierUnaligner#processBarrier and #notifyBarrierReceived
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 64ca88ac989ee7525cb821670f293404b7b30d2d
Author: Zhijiang
AuthorDate: Fri May 29 11:57:33 2020 +0200
[FLINK-17994][checkpointing] Fix the race condition between
CheckpointBarrierUnaligner#processBarrier and #notifyBarrierReceived
The race condition happens as following:
1. CheckpointBarrierUnaligner#notifyBarrierReceived triggers an async
checkpoint(ch1) in mailbox by netty thread.
2. CheckpointBarrierUnaligner#processBarrier also triggers a sync
checkpoint(ch2) by task thread and executes immediately.
3. When ch1 is taken from mailbox by task thread to execute, it will cause
illegal argument exception because it is smaller than the previous executed ch2.
For async checkpoint action, before it is actual executing, we can compare
its id with previous executed checkpoint id. If it is not larger than the
previous
one, we should ignore it to exit directly.
This closes #12406.
---
.../runtime/io/CheckpointBarrierUnaligner.java | 26 --
.../flink/streaming/runtime/tasks/StreamTask.java | 3 +-
.../runtime/io/CheckpointBarrierUnalignerTest.java | 100 -
3 files changed, 119 insertions(+), 10 deletions(-)
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java
index 01b1219..1d4bf82 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java
@@ -137,8 +137,9 @@ public class CheckpointBarrierUnaligner extends
CheckpointBarrierHandler {
}
/**
-* We still need to trigger checkpoint while reading the first barrier
from one channel, because
-* this might happen earlier than the previous async trigger via
mailbox by netty thread.
+* We still need to trigger checkpoint via {@link
ThreadSafeUnaligner#notifyBarrierReceived(CheckpointBarrier, InputChannelInfo)}
+* while reading the first barrier from one channel, because this might
happen
+* earlier than the previous async trigger via mailbox by netty thread.
*
* Note this is also suitable for the trigger case of local input
channel.
*/
@@ -256,8 +257,20 @@ public class CheckpointBarrierUnaligner extends
CheckpointBarrierHandler {
return threadSafeUnaligner.getNumOpenChannels();
}
+ @VisibleForTesting
+ ThreadSafeUnaligner getThreadSafeUnaligner() {
+ return threadSafeUnaligner;
+ }
+
+ private void notifyCheckpoint(CheckpointBarrier barrier) throws
IOException {
+ // ignore the previous triggered checkpoint by netty thread if
it was already canceled or aborted before.
+ if (barrier.getId() >=
threadSafeUnaligner.getCurrentCheckpointId()) {
+ super.notifyCheckpoint(barrier, 0);
+ }
+ }
+
@ThreadSafe
- private static class ThreadSafeUnaligner implements
BufferReceivedListener, Closeable {
+ static class ThreadSafeUnaligner implements BufferReceivedListener,
Closeable {
/**
* Tag the state of which input channel has not received the
barrier, such that newly arriving buffers need
@@ -280,7 +293,6 @@ public class CheckpointBarrierUnaligner extends
CheckpointBarrierHandler {
*/
private long currentReceivedCheckpointId = -1L;
- /** The number of open channels. */
private int numOpenChannels;
private final ChannelStateWriter channelStateWriter;
@@ -300,7 +312,7 @@ public class CheckpointBarrierUnaligner extends
CheckpointBarrierHandler {
if (currentReceivedCheckpointId < barrierId) {
handleNewCheckpoint(barrier);
- handler.executeInTaskThread(() ->
handler.notifyCheckpoint(barrier, 0), "notifyCheckpoint");
+ handler.executeInTaskThread(() ->
handler.notifyCheckpoint(barrier), "notifyCheckpoint");
}
int channelIndex =
handler.getFlattenedChannelIndex(channelInfo);
@@ -396,5 +408,9 @@ public class CheckpointBarrierUnaligner extends
CheckpointBarrierHandler {
synchronized int getNumOpenChannels() {
return numOpenChannels;
}
+
+ synchronized long getCurrentCheckpointId() {
+ return currentReceivedCheckpointId;
+
[flink] branch release-1.11 updated (9737f50 -> 64ca88a)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch release-1.11 in repository https://gitbox.apache.org/repos/asf/flink.git. from 9737f50 [FLINK-17230] Fix incorrect returned address of Endpoint for external Service of ClusterIP type new e25b950 [hotfix][checkpointing] Fix the formatting of CheckpointBarrierUnaligner new 64ca88a [FLINK-17994][checkpointing] Fix the race condition between CheckpointBarrierUnaligner#processBarrier and #notifyBarrierReceived The 2 revisions listed above as "new" are entirely new to this repository and will be described in separate emails. The revisions listed as "add" were already present in the repository and have only been added to this reference. Summary of changes: .../runtime/io/CheckpointBarrierUnaligner.java | 99 ++-- .../flink/streaming/runtime/tasks/StreamTask.java | 3 +- .../runtime/io/CheckpointBarrierUnalignerTest.java | 100 - 3 files changed, 149 insertions(+), 53 deletions(-)
[flink] 01/02: [hotfix][checkpointing] Fix the formatting of CheckpointBarrierUnaligner
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e25b950f2c351c24515e769a175ec8d52e29f835
Author: Zhijiang
AuthorDate: Thu May 28 23:34:27 2020 +0800
[hotfix][checkpointing] Fix the formatting of CheckpointBarrierUnaligner
---
.../runtime/io/CheckpointBarrierUnaligner.java | 81 +-
1 file changed, 34 insertions(+), 47 deletions(-)
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java
index d39accf..01b1219 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java
@@ -130,9 +130,6 @@ public class CheckpointBarrierUnaligner extends
CheckpointBarrierHandler {
/**
* For unaligned checkpoint, it never blocks processing from the task
aspect.
-*
-* For PoC, we do not consider the possibility that the unaligned
checkpoint would
-* not perform due to the max configured unaligned checkpoint size.
*/
@Override
public boolean isBlocked(int channelIndex) {
@@ -140,16 +137,13 @@ public class CheckpointBarrierUnaligner extends
CheckpointBarrierHandler {
}
/**
-* We still need to trigger checkpoint while reading the first barrier
from one channel, because this might happen
-* earlier than the previous async trigger via mailbox by netty thread.
And the {@link AbstractInvokable} has the
-* deduplication logic to guarantee trigger checkpoint only once
finally.
+* We still need to trigger checkpoint while reading the first barrier
from one channel, because
+* this might happen earlier than the previous async trigger via
mailbox by netty thread.
*
* Note this is also suitable for the trigger case of local input
channel.
*/
@Override
- public void processBarrier(
- CheckpointBarrier receivedBarrier,
- int channelIndex) throws Exception {
+ public void processBarrier(CheckpointBarrier receivedBarrier, int
channelIndex) throws Exception {
long barrierId = receivedBarrier.getId();
if (currentConsumedCheckpointId > barrierId ||
(currentConsumedCheckpointId == barrierId && !isCheckpointPending())) {
// ignore old and cancelled barriers
@@ -164,33 +158,32 @@ public class CheckpointBarrierUnaligner extends
CheckpointBarrierHandler {
hasInflightBuffers[channelIndex] = false;
numBarrierConsumed++;
}
- // processBarrier is called from task thread and can actually
happen before notifyBarrierReceived on empty
- // buffer queues
- // to avoid replicating any logic, we simply call
notifyBarrierReceived here as well
threadSafeUnaligner.notifyBarrierReceived(receivedBarrier,
channelInfos[channelIndex]);
}
@Override
public void processCancellationBarrier(CancelCheckpointMarker
cancelBarrier) throws Exception {
- final long barrierId = cancelBarrier.getCheckpointId();
+ long cancelledId = cancelBarrier.getCheckpointId();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("{}: Checkpoint {} canceled, aborting
alignment.", taskName, cancelledId);
+ }
- if (currentConsumedCheckpointId >= barrierId &&
!isCheckpointPending()) {
+ if (currentConsumedCheckpointId >= cancelledId &&
!isCheckpointPending()) {
return;
}
if (isCheckpointPending()) {
LOG.warn("{}: Received cancellation barrier for
checkpoint {} before completing current checkpoint {}. " +
- "Skipping current
checkpoint.",
- taskName,
- barrierId,
- currentConsumedCheckpointId);
- } else if (LOG.isDebugEnabled()) {
- LOG.debug("{}: Checkpoint {} canceled, aborting
alignment.", taskName, barrierId);
+ "Skipping current checkpoint.",
+ taskName,
+ cancelledId,
+ currentConsumedCheckpointId);
}
+
relea
[flink] branch master updated (1c78ab3 -> 8c7c726)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git. from 1c78ab3 [FLINK-17340][docs] Update docs which related to default planner changing. add 8c7c726 [FLINK-17823][network] Resolve the race condition while releasing RemoteInputChannel No new revisions were added by this update. Summary of changes: .../partition/consumer/RemoteInputChannel.java | 15 +-- .../partition/consumer/RemoteInputChannelTest.java | 51 ++ 2 files changed, 63 insertions(+), 3 deletions(-)
[flink] branch master updated (1c78ab3 -> 8c7c726)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git. from 1c78ab3 [FLINK-17340][docs] Update docs which related to default planner changing. add 8c7c726 [FLINK-17823][network] Resolve the race condition while releasing RemoteInputChannel No new revisions were added by this update. Summary of changes: .../partition/consumer/RemoteInputChannel.java | 15 +-- .../partition/consumer/RemoteInputChannelTest.java | 51 ++ 2 files changed, 63 insertions(+), 3 deletions(-)
[flink] branch master updated: [FLINK-17992][checkpointing] Exception from RemoteInputChannel#onBuffer should not fail the whole NetworkClientHandler
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 371f3de [FLINK-17992][checkpointing] Exception from
RemoteInputChannel#onBuffer should not fail the whole NetworkClientHandler
371f3de is described below
commit 371f3de5371afb78d465315098bebec6ed36656b
Author: Zhijiang
AuthorDate: Thu May 28 09:00:34 2020 +0200
[FLINK-17992][checkpointing] Exception from RemoteInputChannel#onBuffer
should not fail the whole NetworkClientHandler
RemoteInputChannel#onBuffer is invoked by
CreditBasedPartitionRequestClientHandler while receiving and decoding the
network data. #onBuffer can
throw exceptions which would tag the error in client handler and fail all
the added input channels inside handler. Then it would cause a tricky
potential issue as following.
If the RemoteInputChannel is canceling by canceler thread, then the task
thread might exit early than canceler thread terminate. That means the
PartitionRequestClient might not be closed (triggered by canceler thread)
while the new task attempt is already deployed into the same TaskManager.
Therefore the new task might reuse the previous PartitionRequestClient
while requesting partitions, but note that the respective client handler was
already tagged an error before during above RemoteInputChannel#onBuffer, to
cause the next round unnecessary failover.
The solution is to only fail the respective task when its internal
RemoteInputChannel#onBuffer throws any exceptions instead of failing the whole
channels inside client handler, then the client is still healthy and can
also be reused by other input channels as long as it is not released yet.
---
.../CreditBasedPartitionRequestClientHandler.java | 11 +++-
...editBasedPartitionRequestClientHandlerTest.java | 71 ++
2 files changed, 80 insertions(+), 2 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java
index 4f3d872..5097c13 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.io.network.netty;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.runtime.io.network.NetworkClientHandler;
import
org.apache.flink.runtime.io.network.netty.exception.LocalTransportException;
import
org.apache.flink.runtime.io.network.netty.exception.RemoteTransportException;
@@ -236,7 +237,8 @@ class CreditBasedPartitionRequestClientHandler extends
ChannelInboundHandlerAdap
/**
* Checks for an error and rethrows it if one was reported.
*/
- private void checkError() throws IOException {
+ @VisibleForTesting
+ void checkError() throws IOException {
final Throwable t = channelError.get();
if (t != null) {
@@ -264,7 +266,12 @@ class CreditBasedPartitionRequestClientHandler extends
ChannelInboundHandlerAdap
return;
}
- decodeBufferOrEvent(inputChannel, bufferOrEvent);
+ try {
+ decodeBufferOrEvent(inputChannel,
bufferOrEvent);
+ } catch (Throwable t) {
+ inputChannel.onError(t);
+ }
+
} else if (msgClazz == NettyMessage.ErrorResponse.class) {
// Error
-
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandlerTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandlerTest.java
index 9487c1c..8cfa4e4 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandlerTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandlerTest.java
@@ -18,8 +18,10 @@
package org.apache.flink.runtime.io.network.netty;
+import org.apache.flink.metrics.SimpleCounter;
import org.apache.flink.runtime.io.network.ConnectionID;
import org.apache.flink.runtime.io.network.PartitionRequestClient;
+import org.apache.flink.runtime.io.network.TestingConnectionManager;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import
[flink] branch master updated (19bbd6d -> aa882c1)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git. from 19bbd6d [FLINK-17945][python] Improve the error message when instantiating non-existing Java class add 9add433 [hotfix][checkpointing] Fix the formatting of CheckpointBarrierUnaligner add aa882c1 [FLINK-17994][checkpointing] Fix the race condition between CheckpointBarrierUnaligner#processBarrier and #notifyBarrierReceived No new revisions were added by this update. Summary of changes: .../runtime/io/CheckpointBarrierUnaligner.java | 99 ++-- .../flink/streaming/runtime/tasks/StreamTask.java | 3 +- .../runtime/io/CheckpointBarrierUnalignerTest.java | 100 - 3 files changed, 149 insertions(+), 53 deletions(-)
[flink] branch master updated: [FLINK-18139][checkpointing] Fixing unaligned checkpoints checks wrong channels for inflight data.
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new d25609d [FLINK-18139][checkpointing] Fixing unaligned checkpoints
checks wrong channels for inflight data.
d25609d is described below
commit d25609d07f52ba9bac3f7bb33ce0635e255e3d9d
Author: Arvid Heise
AuthorDate: Thu Jun 4 22:51:18 2020 +0200
[FLINK-18139][checkpointing] Fixing unaligned checkpoints checks wrong
channels for inflight data.
CheckpointBarrierUnaligner#hasInflightData was not called with input gate
contextual information, such that only the same first few channels are checked
during initial snapshotting of inflight data for multi-gate setups.
---
.../partition/consumer/SingleInputGateBuilder.java | 4 +-
.../io/AlternatingCheckpointBarrierHandler.java| 5 +-
.../runtime/io/CheckpointBarrierHandler.java | 3 +-
.../runtime/io/CheckpointBarrierUnaligner.java | 4 +-
.../runtime/io/CheckpointedInputGate.java | 5 +-
.../AlternatingCheckpointBarrierHandlerTest.java | 4 +-
.../runtime/io/StreamTaskNetworkInputTest.java | 73 +-
7 files changed, 85 insertions(+), 13 deletions(-)
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java
index ad607f7..b279998 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java
@@ -70,12 +70,12 @@ public class SingleInputGateBuilder {
return this;
}
- SingleInputGateBuilder setConsumedSubpartitionIndex(int
consumedSubpartitionIndex) {
+ public SingleInputGateBuilder setConsumedSubpartitionIndex(int
consumedSubpartitionIndex) {
this.consumedSubpartitionIndex = consumedSubpartitionIndex;
return this;
}
- SingleInputGateBuilder setSingleInputGateIndex(int gateIndex) {
+ public SingleInputGateBuilder setSingleInputGateIndex(int gateIndex) {
this.gateIndex = gateIndex;
return this;
}
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/AlternatingCheckpointBarrierHandler.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/AlternatingCheckpointBarrierHandler.java
index 09f05b9..8ed6788 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/AlternatingCheckpointBarrierHandler.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/AlternatingCheckpointBarrierHandler.java
@@ -18,6 +18,7 @@
package org.apache.flink.streaming.runtime.io;
import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo;
import org.apache.flink.runtime.io.network.api.CancelCheckpointMarker;
import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
import org.apache.flink.runtime.io.network.buffer.BufferReceivedListener;
@@ -99,9 +100,9 @@ class AlternatingCheckpointBarrierHandler extends
CheckpointBarrierHandler {
}
@Override
- public boolean hasInflightData(long checkpointId, int channelIndex) {
+ public boolean hasInflightData(long checkpointId, InputChannelInfo
channelInfo) {
// should only be called for unaligned checkpoint
- return unalignedHandler.hasInflightData(checkpointId,
channelIndex);
+ return unalignedHandler.hasInflightData(checkpointId,
channelInfo);
}
@Override
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierHandler.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierHandler.java
index 07f6f7a..952af24 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierHandler.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierHandler.java
@@ -22,6 +22,7 @@ import
org.apache.flink.runtime.checkpoint.CheckpointException;
import org.apache.flink.runtime.checkpoint.CheckpointFailureReason;
import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
+import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo;
import org.apache.flink.runtime.io.network.api.CancelCheckpointMarker;
import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
import
[flink] branch release-1.11 updated: [FLINK-18139][checkpointing] Fixing unaligned checkpoints checks wrong channels for inflight data.
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.11 by this push:
new b2852d3 [FLINK-18139][checkpointing] Fixing unaligned checkpoints
checks wrong channels for inflight data.
b2852d3 is described below
commit b2852d3f80ea767fb730290ee49706280559ffc8
Author: Arvid Heise
AuthorDate: Thu Jun 4 22:51:18 2020 +0200
[FLINK-18139][checkpointing] Fixing unaligned checkpoints checks wrong
channels for inflight data.
CheckpointBarrierUnaligner#hasInflightData was not called with input gate
contextual information, such that only the same first few channels are checked
during initial snapshotting of inflight data for multi-gate setups.
---
.../partition/consumer/SingleInputGateBuilder.java | 4 +-
.../io/AlternatingCheckpointBarrierHandler.java| 5 +-
.../runtime/io/CheckpointBarrierHandler.java | 3 +-
.../runtime/io/CheckpointBarrierUnaligner.java | 4 +-
.../runtime/io/CheckpointedInputGate.java | 5 +-
.../AlternatingCheckpointBarrierHandlerTest.java | 4 +-
.../runtime/io/StreamTaskNetworkInputTest.java | 73 +-
7 files changed, 85 insertions(+), 13 deletions(-)
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java
index ad607f7..b279998 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java
@@ -70,12 +70,12 @@ public class SingleInputGateBuilder {
return this;
}
- SingleInputGateBuilder setConsumedSubpartitionIndex(int
consumedSubpartitionIndex) {
+ public SingleInputGateBuilder setConsumedSubpartitionIndex(int
consumedSubpartitionIndex) {
this.consumedSubpartitionIndex = consumedSubpartitionIndex;
return this;
}
- SingleInputGateBuilder setSingleInputGateIndex(int gateIndex) {
+ public SingleInputGateBuilder setSingleInputGateIndex(int gateIndex) {
this.gateIndex = gateIndex;
return this;
}
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/AlternatingCheckpointBarrierHandler.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/AlternatingCheckpointBarrierHandler.java
index 09f05b9..8ed6788 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/AlternatingCheckpointBarrierHandler.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/AlternatingCheckpointBarrierHandler.java
@@ -18,6 +18,7 @@
package org.apache.flink.streaming.runtime.io;
import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo;
import org.apache.flink.runtime.io.network.api.CancelCheckpointMarker;
import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
import org.apache.flink.runtime.io.network.buffer.BufferReceivedListener;
@@ -99,9 +100,9 @@ class AlternatingCheckpointBarrierHandler extends
CheckpointBarrierHandler {
}
@Override
- public boolean hasInflightData(long checkpointId, int channelIndex) {
+ public boolean hasInflightData(long checkpointId, InputChannelInfo
channelInfo) {
// should only be called for unaligned checkpoint
- return unalignedHandler.hasInflightData(checkpointId,
channelIndex);
+ return unalignedHandler.hasInflightData(checkpointId,
channelInfo);
}
@Override
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierHandler.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierHandler.java
index 07f6f7a..952af24 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierHandler.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierHandler.java
@@ -22,6 +22,7 @@ import
org.apache.flink.runtime.checkpoint.CheckpointException;
import org.apache.flink.runtime.checkpoint.CheckpointFailureReason;
import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
+import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo;
import org.apache.flink.runtime.io.network.api.CancelCheckpointMarker;
import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
import
[flink] branch release-1.11 updated (cc59b3a -> b2852d3)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch release-1.11 in repository https://gitbox.apache.org/repos/asf/flink.git. from cc59b3a [FLINK-18020] Increase timeout in SQLClientKafkaITCase add b2852d3 [FLINK-18139][checkpointing] Fixing unaligned checkpoints checks wrong channels for inflight data. No new revisions were added by this update. Summary of changes: .../partition/consumer/SingleInputGateBuilder.java | 4 +- .../io/AlternatingCheckpointBarrierHandler.java| 5 +- .../runtime/io/CheckpointBarrierHandler.java | 3 +- .../runtime/io/CheckpointBarrierUnaligner.java | 4 +- .../runtime/io/CheckpointedInputGate.java | 5 +- .../AlternatingCheckpointBarrierHandlerTest.java | 4 +- .../runtime/io/StreamTaskNetworkInputTest.java | 73 +- 7 files changed, 85 insertions(+), 13 deletions(-)
[flink] branch master updated (44af789 -> f2dd4b8)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git. from 44af789 [FLINK-16350] Support Zookeeper 3.5 in test_ha_per_job_cluster_datastream.sh add ed7b0b1 [FLINK-18050][task][checkpointing] Use CloseableIterator to write ResultSubpartition state add f2dd4b8 [FLINK-18050][task][checkpointing] Simplify ChannelStateCheckpointWriter interface No new revisions were added by this update. Summary of changes: .../channel/ChannelStateCheckpointWriter.java | 16 +++-- .../channel/ChannelStateWriteRequest.java | 22 +-- .../checkpoint/channel/ChannelStateWriterImpl.java | 23 +--- ...ChannelStateWriteRequestDispatcherImplTest.java | 69 ++ .../ChannelStateWriteRequestDispatcherTest.java| 2 +- 5 files changed, 94 insertions(+), 38 deletions(-) create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriteRequestDispatcherImplTest.java
[flink] branch release-1.11 updated (ab6cf40 -> 822e01b)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch release-1.11 in repository https://gitbox.apache.org/repos/asf/flink.git. from ab6cf40 [FLINK-18130][hive][fs-connector] File name conflict for different jobs in filesystem/hive sink () add a233c0f [FLINK-18050][task][checkpointing] Use CloseableIterator to write ResultSubpartition state add 822e01b [FLINK-18050][task][checkpointing] Simplify ChannelStateCheckpointWriter interface No new revisions were added by this update. Summary of changes: .../channel/ChannelStateCheckpointWriter.java | 16 +++-- .../channel/ChannelStateWriteRequest.java | 22 +-- .../checkpoint/channel/ChannelStateWriterImpl.java | 23 +--- ...ChannelStateWriteRequestDispatcherImplTest.java | 69 ++ .../ChannelStateWriteRequestDispatcherTest.java| 2 +- 5 files changed, 94 insertions(+), 38 deletions(-) create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriteRequestDispatcherImplTest.java
[flink] branch master updated (64f8f65 -> 6dc624d)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git. from 64f8f65 [FLINK-18042][tests] Auto-detect distDir add 6dc624d [FLINK-18136][checkpointing] Don't start channel state writer for savepoint No new revisions were added by this update. Summary of changes: .../state/TestCheckpointStorageWorkerView.java | 52 + .../runtime/io/CheckpointBarrierUnaligner.java | 15 ++-- .../streaming/runtime/io/InputProcessorUtil.java | 14 ++-- .../runtime/tasks/MultipleInputStreamTask.java | 2 +- .../runtime/tasks/OneInputStreamTask.java | 2 +- .../flink/streaming/runtime/tasks/StreamTask.java | 7 +- .../tasks/SubtaskCheckpointCoordinator.java| 10 ++- .../tasks/SubtaskCheckpointCoordinatorImpl.java| 50 ++-- .../runtime/tasks/TwoInputStreamTask.java | 2 +- .../AlternatingCheckpointBarrierHandlerTest.java | 12 +-- ...CheckpointBarrierUnalignerCancellationTest.java | 4 +- .../runtime/io/CheckpointBarrierUnalignerTest.java | 6 +- .../runtime/io/InputProcessorUtilTest.java | 3 +- .../runtime/io/StreamTaskNetworkInputTest.java | 5 +- .../tasks/SubtaskCheckpointCoordinatorTest.java| 47 .../tasks/TestSubtaskCheckpointCoordinator.java| 89 ++ 16 files changed, 278 insertions(+), 42 deletions(-) create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/state/TestCheckpointStorageWorkerView.java create mode 100644 flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TestSubtaskCheckpointCoordinator.java
[flink] branch release-1.11 updated (822e8a3 -> 2fe888c)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch release-1.11 in repository https://gitbox.apache.org/repos/asf/flink.git. from 822e8a3 [FLINK-18152][Depl] Fail fast in JM scripts if memory configuration failed add b552ff2 [hotfix][checkpointing] Add VisibleForTesting annotation for related methods add 9860579 [FLINK-18063][checkpointing][refactoring] Implement default #isBlocked method in CheckpointBarrierHandler add 587a399 [FLINK-18063][checkpointing] Fix the invalid implementation of AlternatingCheckpointBarrierHandler#getAlignmentDurationNanos add 2fe888c [FLINK-18063][checkpointing] Fix the race condition of aborting checkpoint in CheckpointBarrierUnaligner No new revisions were added by this update. Summary of changes: .../io/AlternatingCheckpointBarrierHandler.java| 21 ++-- .../runtime/io/CheckpointBarrierAligner.java | 8 ++ .../runtime/io/CheckpointBarrierHandler.java | 14 ++- .../runtime/io/CheckpointBarrierTracker.java | 14 --- .../runtime/io/CheckpointBarrierUnaligner.java | 126 ++-- .../runtime/io/CheckpointedInputGate.java | 3 +- ...CheckpointBarrierUnalignerCancellationTest.java | 2 +- .../runtime/io/CheckpointBarrierUnalignerTest.java | 129 - 8 files changed, 222 insertions(+), 95 deletions(-)
[flink] branch master updated (44acd50 -> f672ae5)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git. from 44acd50 [FLINK-18152][Depl] Fail fast in JM scripts if memory configuration failed add bb0ebca [hotfix][checkpointing] Add VisibleForTesting annotation for related methods add 0cbd5e3 [FLINK-18063][checkpointing][refactoring] Implement default #isBlocked method in CheckpointBarrierHandler add d99e1f8 [FLINK-18063][checkpointing] Fix the invalid implementation of AlternatingCheckpointBarrierHandler#getAlignmentDurationNanos add f672ae5 [FLINK-18063][checkpointing] Fix the race condition of aborting checkpoint in CheckpointBarrierUnaligner No new revisions were added by this update. Summary of changes: .../io/AlternatingCheckpointBarrierHandler.java| 21 ++-- .../runtime/io/CheckpointBarrierAligner.java | 8 ++ .../runtime/io/CheckpointBarrierHandler.java | 14 ++- .../runtime/io/CheckpointBarrierTracker.java | 14 --- .../runtime/io/CheckpointBarrierUnaligner.java | 126 ++-- .../runtime/io/CheckpointedInputGate.java | 3 +- ...CheckpointBarrierUnalignerCancellationTest.java | 2 +- .../runtime/io/CheckpointBarrierUnalignerTest.java | 129 - 8 files changed, 222 insertions(+), 95 deletions(-)
[flink] branch master updated (a80dc5a -> 892e1c3)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git. from a80dc5a [hotfix][tests] Add missing import add 892e1c3 [hotfix][tests] Fix the compile issue caused by constructor change of CheckpointBarrierUnaligner No new revisions were added by this update. Summary of changes: .../streaming/runtime/io/CheckpointBarrierUnalignerTest.java | 8 1 file changed, 4 insertions(+), 4 deletions(-)
[flink] branch master updated (892e1c3 -> 131cf30)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git. from 892e1c3 [hotfix][tests] Fix the compile issue caused by constructor change of CheckpointBarrierUnaligner add 131cf30 [hotfix][testing] Remove the unused import from CheckpointBarrierUnalignerTest No new revisions were added by this update. Summary of changes: .../flink/streaming/runtime/io/CheckpointBarrierUnalignerTest.java | 1 - 1 file changed, 1 deletion(-)
[flink] branch release-1.11 updated (2fb7b14 -> fe229a3)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch release-1.11 in repository https://gitbox.apache.org/repos/asf/flink.git. from 2fb7b14 [hotfix] Improve JavaDocs comments for FlinkMemory/FlinkMemoryUtis add fe229a3 [FLINK-18057][tests] Fix unstable test SingleInputGateTest#testConcurrentReadStateAndProcessAndClose No new revisions were added by this update. Summary of changes: .../runtime/io/network/partition/consumer/SingleInputGateTest.java | 6 +- 1 file changed, 5 insertions(+), 1 deletion(-)
[flink] branch master updated (d757fc1 -> 9a7dbfc)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git. from d757fc1 [hotfix] Improve JavaDocs comments for FlinkMemory/FlinkMemoryUtis add 9a7dbfc [FLINK-18057][tests] Fix unstable test SingleInputGateTest#testConcurrentReadStateAndProcessAndClose No new revisions were added by this update. Summary of changes: .../runtime/io/network/partition/consumer/SingleInputGateTest.java | 6 +- 1 file changed, 5 insertions(+), 1 deletion(-)
[flink] branch release-1.11 updated: [FLINK-18136][checkpointing] Don't start channel state writer for savepoint
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.11 by this push:
new 7f410fa [FLINK-18136][checkpointing] Don't start channel state writer
for savepoint
7f410fa is described below
commit 7f410fa4f4758e48fcc6c33d8354c21ab02e2dc6
Author: Roman
AuthorDate: Tue Jun 9 10:58:46 2020 +0200
[FLINK-18136][checkpointing] Don't start channel state writer for savepoint
ChannelStateWriter#start should be only called for unaligned checkpoint.
While source triggering
savepoint, SubtaskCheckpointCoordinator#initCheckpoint is introduced to
judge the condition
whether to start the internal writer or not. And this new method is also
used in other places like
CheckpointBarrierUnaligner.
This closes #12489.
---
.../state/TestCheckpointStorageWorkerView.java | 52 +
.../runtime/io/CheckpointBarrierUnaligner.java | 15 ++--
.../streaming/runtime/io/InputProcessorUtil.java | 14 ++--
.../runtime/tasks/MultipleInputStreamTask.java | 2 +-
.../runtime/tasks/OneInputStreamTask.java | 2 +-
.../flink/streaming/runtime/tasks/StreamTask.java | 7 +-
.../tasks/SubtaskCheckpointCoordinator.java| 10 ++-
.../tasks/SubtaskCheckpointCoordinatorImpl.java| 50 ++--
.../runtime/tasks/TwoInputStreamTask.java | 2 +-
.../AlternatingCheckpointBarrierHandlerTest.java | 12 +--
...CheckpointBarrierUnalignerCancellationTest.java | 4 +-
.../runtime/io/CheckpointBarrierUnalignerTest.java | 14 ++--
.../runtime/io/InputProcessorUtilTest.java | 3 +-
.../runtime/io/StreamTaskNetworkInputTest.java | 5 +-
.../tasks/SubtaskCheckpointCoordinatorTest.java| 47
.../tasks/TestSubtaskCheckpointCoordinator.java| 89 ++
16 files changed, 282 insertions(+), 46 deletions(-)
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/state/TestCheckpointStorageWorkerView.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/state/TestCheckpointStorageWorkerView.java
new file mode 100644
index 000..c50a390
--- /dev/null
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/state/TestCheckpointStorageWorkerView.java
@@ -0,0 +1,52 @@
+/*
+ * 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.state;
+
+import org.apache.flink.runtime.state.memory.MemCheckpointStreamFactory;
+
+import java.io.IOException;
+
+/**
+ * Non-persistent {@link CheckpointStorageWorkerView} for tests. Uses {@link
MemCheckpointStreamFactory}.
+ */
+public class TestCheckpointStorageWorkerView implements
CheckpointStorageWorkerView {
+
+ private final int maxStateSize;
+ private final MemCheckpointStreamFactory
taskOwnedCheckpointStreamFactory;
+ private final CheckpointedStateScope taskOwnedStateScope;
+
+ public TestCheckpointStorageWorkerView(int maxStateSize) {
+ this(maxStateSize, CheckpointedStateScope.EXCLUSIVE);
+ }
+
+ private TestCheckpointStorageWorkerView(int maxStateSize,
CheckpointedStateScope taskOwnedStateScope) {
+ this.maxStateSize = maxStateSize;
+ this.taskOwnedCheckpointStreamFactory = new
MemCheckpointStreamFactory(maxStateSize);
+ this.taskOwnedStateScope = taskOwnedStateScope;
+ }
+
+ @Override
+ public CheckpointStreamFactory resolveCheckpointStorageLocation(long
checkpointId, CheckpointStorageLocationReference reference) {
+ return new MemCheckpointStreamFactory(maxStateSize);
+ }
+
+ @Override
+ public CheckpointStreamFactory.CheckpointStateOutputStream
createTaskOwnedStateStream() throws IOException {
+ return
taskOwnedCheckpointStreamFactory.createCheckpointStateOutputStream(taskOwnedStateScope);
+ }
+}
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/Checkpo
[flink] branch master updated (7143e6a -> b2d1c2d)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git. from 7143e6a [FLINK-17944][sql-client] Wrong output in SQL Client's table mode add 1cb9d54 [FLINK-17869][hotfix] Add taskName to ChannelStateWriter log messages add 82a13db [FLINK-17869][hotfix] Don't pass ChannelStateWrite Future to AsyncCheckpointRunnable add 73afe2b [FLINK-17869][task][checkpointing] Revert "[FLINK-17218][checkpointing] Ensuring that ChannelStateWriter aborts previous checkpoints before a new checkpoint is started." add 91df1a5 [FLINK-17869][task][checkpointing] Abort channel state write if checkpoint is subsumed add 2b51cda [FLINK-17869][task][checkpointing] Increase ChannelStateWriterImpl.DEFAULT_MAX_CHECKPOINTS add 64ff676 [FLINK-17869][task][checkpointing] Ignore out of order checkpoints in SubtaskCheckpointCoordinator add 26762bd1 [FLINK-17869][tests] Unignore UnalignedCheckpointITCase add b2d1c2d [FLINK-17869][task][checkpointing] Abort writing of channel state by RPC notification No new revisions were added by this update. Summary of changes: .../checkpoint/channel/ChannelStateWriter.java | 29 +--- .../checkpoint/channel/ChannelStateWriterImpl.java | 25 -- .../channel/ChannelStateWriterImplTest.java| 49 +--- .../checkpoint/channel/MockChannelStateWriter.java | 8 +--- .../channel/RecordingChannelStateWriter.java | 5 -- .../runtime/state/ChannelPersistenceITCase.java| 2 +- .../runtime/io/CheckpointBarrierUnaligner.java | 31 +++-- .../runtime/tasks/AsyncCheckpointRunnable.java | 6 --- .../tasks/SubtaskCheckpointCoordinatorImpl.java| 54 +- .../runtime/tasks/LocalStateForwardingTest.java| 2 - .../tasks/TestSubtaskCheckpointCoordinator.java| 2 +- .../checkpointing/UnalignedCheckpointITCase.java | 2 - 12 files changed, 95 insertions(+), 120 deletions(-)
[flink] 01/08: [FLINK-17869][hotfix] Add taskName to ChannelStateWriter log messages
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4030b1b30a2208e494d639af8767c3a944d87d58
Author: Roman Khachatryan
AuthorDate: Wed Jun 3 14:34:30 2020 +0200
[FLINK-17869][hotfix] Add taskName to ChannelStateWriter log messages
---
.../flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java| 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
index e6aa9dc..fc8655c 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
@@ -102,11 +102,11 @@ public class ChannelStateWriterImpl implements
ChannelStateWriter {
LOG.debug("{} starting checkpoint {} ({})", taskName,
checkpointId, checkpointOptions);
ChannelStateWriteResult result = new ChannelStateWriteResult();
ChannelStateWriteResult put =
results.computeIfAbsent(checkpointId, id -> {
- Preconditions.checkState(results.size() <
maxCheckpoints, "results.size() > maxCheckpoints", results.size(),
maxCheckpoints);
+ Preconditions.checkState(results.size() <
maxCheckpoints, String.format("%s can't start %d, results.size() >
maxCheckpoints: %d > %d", taskName, checkpointId, results.size(),
maxCheckpoints));
enqueue(new CheckpointStartRequest(checkpointId,
result, checkpointOptions.getTargetLocation()), false);
return result;
});
- Preconditions.checkArgument(put == result, "result future
already present for checkpoint " + checkpointId);
+ Preconditions.checkArgument(put == result, taskName + " result
future already present for checkpoint " + checkpointId);
}
@Override
@@ -156,7 +156,7 @@ public class ChannelStateWriterImpl implements
ChannelStateWriter {
public ChannelStateWriteResult getWriteResult(long checkpointId) {
LOG.debug("{} requested write result, checkpoint {}", taskName,
checkpointId);
ChannelStateWriteResult result = results.get(checkpointId);
- Preconditions.checkArgument(result != null, "channel state
write result not found for checkpoint " + checkpointId);
+ Preconditions.checkArgument(result != null, taskName + "
channel state write result not found for checkpoint " + checkpointId);
return result;
}
[flink] 02/08: [FLINK-17869][hotfix] Don't pass ChannelStateWrite Future to AsyncCheckpointRunnable
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 36609bb5dc2aee4061a47f7a767630f1f5912d96
Author: Roman Khachatryan
AuthorDate: Wed Jun 3 23:03:52 2020 +0200
[FLINK-17869][hotfix] Don't pass ChannelStateWrite Future to
AsyncCheckpointRunnable
OperatorSnapshotFinalizer already waits and holds this future.
ChannelStateWriter.getWriteResult() can then be non-idempotent.
ChannelStateWriter.stop() can then be removed.
---
.../checkpoint/channel/ChannelStateWriter.java | 21 ++--
.../checkpoint/channel/ChannelStateWriterImpl.java | 10 ++--
.../channel/ChannelStateWriterImplTest.java| 28 ++
.../checkpoint/channel/MockChannelStateWriter.java | 6 +
.../channel/RecordingChannelStateWriter.java | 5
.../runtime/state/ChannelPersistenceITCase.java| 2 +-
.../runtime/tasks/AsyncCheckpointRunnable.java | 6 -
.../tasks/SubtaskCheckpointCoordinatorImpl.java| 15 +---
.../runtime/tasks/LocalStateForwardingTest.java| 2 --
9 files changed, 24 insertions(+), 71 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
index 5dad559..af2a708 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
@@ -123,7 +123,7 @@ public interface ChannelStateWriter extends Closeable {
* Finalize write of channel state data for the given checkpoint id.
* Must be called after {@link #start(long, CheckpointOptions)} and all
of the input data of the given checkpoint added.
* When both {@link #finishInput} and {@link #finishOutput} were called
the results can be (eventually) obtained
-* using {@link #getWriteResult}
+* using {@link #getAndRemoveWriteResult}
*/
void finishInput(long checkpointId);
@@ -131,24 +131,21 @@ public interface ChannelStateWriter extends Closeable {
* Finalize write of channel state data for the given checkpoint id.
* Must be called after {@link #start(long, CheckpointOptions)} and all
of the output data of the given checkpoint added.
* When both {@link #finishInput} and {@link #finishOutput} were called
the results can be (eventually) obtained
-* using {@link #getWriteResult}
+* using {@link #getAndRemoveWriteResult}
*/
void finishOutput(long checkpointId);
/**
* Aborts the checkpoint and fails pending result for this checkpoint.
+* @param cleanup true if {@link #getAndRemoveWriteResult(long)} is not
supposed to be called afterwards.
*/
void abort(long checkpointId, Throwable cause);
/**
-* Must be called after {@link #start(long, CheckpointOptions)}.
+* Must be called after {@link #start(long, CheckpointOptions)} once.
+* @throws IllegalArgumentException if the passed checkpointId is not
known.
*/
- ChannelStateWriteResult getWriteResult(long checkpointId);
-
- /**
-* Cleans up the internal state for the given checkpoint.
-*/
- void stop(long checkpointId);
+ ChannelStateWriteResult getAndRemoveWriteResult(long checkpointId)
throws IllegalArgumentException;
ChannelStateWriter NO_OP = new NoOpChannelStateWriter();
@@ -181,16 +178,12 @@ public interface ChannelStateWriter extends Closeable {
}
@Override
- public ChannelStateWriteResult getWriteResult(long
checkpointId) {
+ public ChannelStateWriteResult getAndRemoveWriteResult(long
checkpointId) {
return ChannelStateWriteResult.EMPTY;
}
@Override
public void close() {
}
-
- @Override
- public void stop(long checkpointId) {
- }
}
}
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
index fc8655c..6158358 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
@@ -153,19 +153,13 @@ public class ChannelStateWriterImpl implements
ChannelStateWriter {
}
@Override
- public ChannelStateWriteResult getWriteResult(long checkpointId) {
+ public ChannelStateWriteR
[flink] 05/08: [FLINK-17869][task][checkpointing] Increase ChannelStateWriterImpl.DEFAULT_MAX_CHECKPOINTS
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e14958dcd09f7c32ba2ba25fc6a48696fe7eeca2
Author: Roman Khachatryan
AuthorDate: Wed Jun 3 15:56:47 2020 +0200
[FLINK-17869][task][checkpointing] Increase
ChannelStateWriterImpl.DEFAULT_MAX_CHECKPOINTS
ChannelStateWriter map is cleaned up by the task thread,
so the check in netty thread should take possible delay
into account.
---
.../apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
index 3e18050..89a5247 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
@@ -56,7 +56,7 @@ import static
org.apache.flink.runtime.checkpoint.channel.ChannelStateWriteReque
public class ChannelStateWriterImpl implements ChannelStateWriter {
private static final Logger LOG =
LoggerFactory.getLogger(ChannelStateWriterImpl.class);
- private static final int DEFAULT_MAX_CHECKPOINTS = 5; // currently,
only single in-flight checkpoint is supported
+ private static final int DEFAULT_MAX_CHECKPOINTS = 1000; // includes
max-concurrent-checkpoints + checkpoints to be aborted (scheduled via mailbox)
private final String taskName;
private final ChannelStateWriteRequestExecutor executor;
[flink] 08/08: [FLINK-17869][task][checkpointing] Abort writing of channel state by RPC notification
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 1d75ec8efe3c31f2a5e1a4572b5cfb8000ddbf67
Author: Roman Khachatryan
AuthorDate: Thu Jun 4 10:06:15 2020 +0200
[FLINK-17869][task][checkpointing] Abort writing of channel state by RPC
notification
---
.../flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java | 2 ++
1 file changed, 2 insertions(+)
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
index 5d7a2c9..d904c87 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
@@ -306,6 +306,8 @@ class SubtaskCheckpointCoordinatorImpl implements
SubtaskCheckpointCoordinator {
}
}
+ channelStateWriter.abort(checkpointId, new
CancellationException("checkpoint aborted via notification"), false);
+
for (StreamOperatorWrapper operatorWrapper :
operatorChain.getAllOperators(true)) {
try {
operatorWrapper.getStreamOperator().notifyCheckpointAborted(checkpointId);
[flink] branch release-1.11 updated (31a17cb -> 1d75ec8)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch release-1.11 in repository https://gitbox.apache.org/repos/asf/flink.git. from 31a17cb [FLINK-18188][Runtime] Derive JM Off-Heap memory from configured Total Flink Memory minus JVM Heap new 4030b1b [FLINK-17869][hotfix] Add taskName to ChannelStateWriter log messages new 36609bb [FLINK-17869][hotfix] Don't pass ChannelStateWrite Future to AsyncCheckpointRunnable new 6cb8f28 [FLINK-17869][task][checkpointing] Revert "[FLINK-17218][checkpointing] Ensuring that ChannelStateWriter aborts previous checkpoints before a new checkpoint is started." new d806924 [FLINK-17869][task][checkpointing] Abort channel state write if checkpoint is subsumed new e14958d [FLINK-17869][task][checkpointing] Increase ChannelStateWriterImpl.DEFAULT_MAX_CHECKPOINTS new 7bb3ffa [FLINK-17869][task][checkpointing] Ignore out of order checkpoints in SubtaskCheckpointCoordinator new 0634a2e [FLINK-17869][tests] Unignore UnalignedCheckpointITCase new 1d75ec8 [FLINK-17869][task][checkpointing] Abort writing of channel state by RPC notification The 8 revisions listed above as "new" are entirely new to this repository and will be described in separate emails. The revisions listed as "add" were already present in the repository and have only been added to this reference. Summary of changes: .../checkpoint/channel/ChannelStateWriter.java | 29 +--- .../checkpoint/channel/ChannelStateWriterImpl.java | 25 -- .../channel/ChannelStateWriterImplTest.java| 49 +--- .../checkpoint/channel/MockChannelStateWriter.java | 8 +--- .../channel/RecordingChannelStateWriter.java | 5 -- .../runtime/state/ChannelPersistenceITCase.java| 2 +- .../runtime/io/CheckpointBarrierUnaligner.java | 31 +++-- .../runtime/tasks/AsyncCheckpointRunnable.java | 6 --- .../tasks/SubtaskCheckpointCoordinatorImpl.java| 54 +- .../runtime/tasks/LocalStateForwardingTest.java| 2 - .../tasks/TestSubtaskCheckpointCoordinator.java| 2 +- .../checkpointing/UnalignedCheckpointITCase.java | 2 - 12 files changed, 95 insertions(+), 120 deletions(-)
[flink] 03/08: [FLINK-17869][task][checkpointing] Revert "[FLINK-17218][checkpointing] Ensuring that ChannelStateWriter aborts previous checkpoints before a new checkpoint is started."
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 6cb8f28cf219f79a36b90a567692fafc5f85a2a2
Author: Roman Khachatryan
AuthorDate: Wed Jun 3 14:12:04 2020 +0200
[FLINK-17869][task][checkpointing] Revert "[FLINK-17218][checkpointing]
Ensuring that ChannelStateWriter aborts previous checkpoints before a new
checkpoint is started."
This reverts commit 24ff415f1b76392f75dea7c3538558d24fcb7058
which introduced a race condition when task thread and netty
thread compete for ChannelStateWriteResult.
Instead, next commits fix it by:
1. Map size validation error will be prevented simply by increasing the
limit
2. When a checkpoint is subsumed, it's write result will be removed from on
future completion
---
.../checkpoint/channel/ChannelStateWriter.java| 4 +++-
.../checkpoint/channel/ChannelStateWriterImpl.java| 1 -
.../channel/ChannelStateWriterImplTest.java | 19 ---
3 files changed, 11 insertions(+), 13 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
index af2a708..02a3a69 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
@@ -179,7 +179,9 @@ public interface ChannelStateWriter extends Closeable {
@Override
public ChannelStateWriteResult getAndRemoveWriteResult(long
checkpointId) {
- return ChannelStateWriteResult.EMPTY;
+ return new ChannelStateWriteResult(
+
CompletableFuture.completedFuture(Collections.emptyList()),
+
CompletableFuture.completedFuture(Collections.emptyList()));
}
@Override
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
index 6158358..8996b3b 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
@@ -98,7 +98,6 @@ public class ChannelStateWriterImpl implements
ChannelStateWriter {
@Override
public void start(long checkpointId, CheckpointOptions
checkpointOptions) {
- results.keySet().forEach(oldCheckpointId ->
abort(oldCheckpointId, new Exception("Starting new checkpoint " +
checkpointId)));
LOG.debug("{} starting checkpoint {} ({})", taskName,
checkpointId, checkpointOptions);
ChannelStateWriteResult result = new ChannelStateWriteResult();
ChannelStateWriteResult put =
results.computeIfAbsent(checkpointId, id -> {
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
index 9d7a7ea..0dae88e 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
@@ -186,19 +186,16 @@ public class ChannelStateWriterImplTest {
unwrappingError(TestException.class, () -> callStart(writer));
}
- @Test
- public void testStartAbortsOldCheckpoints() throws Exception {
- int maxCheckpoints = 10;
- runWithSyncWorker((writer, worker) -> {
- writer.start(0,
CheckpointOptions.forCheckpointWithDefaultLocation());
- ChannelStateWriteResult writeResult =
writer.getWriteResult(0);
- for (int i = 1; i <= maxCheckpoints; i++) {
+ @Test(expected = IllegalStateException.class)
+ public void testLimit() throws IOException {
+ int maxCheckpoints = 3;
+ try (ChannelStateWriterImpl writer = new
ChannelStateWriterImpl(TASK_NAME, getStreamFactoryFactory(), maxCheckpoints)) {
+ writer.open();
+ for (int i = 0; i < maxCheckpoints; i++) {
writer.start(i,
CheckpointOptions.forCheckpointWithDefaultLocation());
- worker.processAllRequests();
- assertTrue(writeResult.isDone());
-
[flink] 06/08: [FLINK-17869][task][checkpointing] Ignore out of order checkpoints in SubtaskCheckpointCoordinator
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 7bb3ffa91a9916348d2f0a6a2e6cba4b109be56e
Author: Roman Khachatryan
AuthorDate: Wed Jun 3 21:43:56 2020 +0200
[FLINK-17869][task][checkpointing] Ignore out of order checkpoints in
SubtaskCheckpointCoordinator
Check (by task thread) whether the current checkpoint was already aborted
in the following scenario:
1. on checkpoint barrier ThreadSafeUnaligner sends a mail to start
checkpointing (netty thread)
2. on cancellation marker CheckpointBarrierUnaligner aborts it (task thread)
3. task thread processes a mail to start checkpointing
---
.../tasks/SubtaskCheckpointCoordinatorImpl.java| 22 +++---
1 file changed, 19 insertions(+), 3 deletions(-)
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
index cf8a21e..5d7a2c9 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
@@ -44,7 +44,6 @@ import
org.apache.flink.streaming.api.operators.StreamOperator;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.IOUtils;
-import org.apache.flink.util.Preconditions;
import org.apache.flink.util.function.BiFunctionWithException;
import org.slf4j.Logger;
@@ -57,10 +56,12 @@ import java.io.UncheckedIOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
+import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
@@ -183,6 +184,16 @@ class SubtaskCheckpointCoordinatorImpl implements
SubtaskCheckpointCoordinator {
@Override
public void abortCheckpointOnBarrier(long checkpointId, Throwable
cause, OperatorChain operatorChain) throws IOException {
LOG.debug("Aborting checkpoint via cancel-barrier {} for task
{}", checkpointId, taskName);
+ lastCheckpointId = Math.max(lastCheckpointId, checkpointId);
+ Iterator iterator = abortedCheckpointIds.iterator();
+ while (iterator.hasNext()) {
+ long next = iterator.next();
+ if (next < lastCheckpointId) {
+ iterator.remove();
+ } else {
+ break;
+ }
+ }
checkpointStorage.clearCacheFor(checkpointId);
@@ -221,9 +232,14 @@ class SubtaskCheckpointCoordinatorImpl implements
SubtaskCheckpointCoordinator {
// We generally try to emit the checkpoint barrier as soon as
possible to not affect downstream
// checkpoint alignments
+ if (lastCheckpointId >= metadata.getCheckpointId()) {
+ LOG.info("Out of order checkpoint barrier (aborted
previously?): {} >= {}", lastCheckpointId, metadata.getCheckpointId());
+ channelStateWriter.abort(metadata.getCheckpointId(),
new CancellationException(), true);
+ checkAndClearAbortedStatus(metadata.getCheckpointId());
+ return;
+ }
+
// Step (0): Record the last triggered checkpointId.
- Preconditions.checkArgument(lastCheckpointId <
metadata.getCheckpointId(), String.format(
- "Unexpected current checkpoint-id: %s vs last
checkpoint-id: %s", metadata.getCheckpointId(), lastCheckpointId));
lastCheckpointId = metadata.getCheckpointId();
if (checkAndClearAbortedStatus(metadata.getCheckpointId())) {
LOG.info("Checkpoint {} has been notified as aborted,
would not trigger any checkpoint.", metadata.getCheckpointId());
[flink] 07/08: [FLINK-17869][tests] Unignore UnalignedCheckpointITCase
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 0634a2efd2f0ca42d4cb6762015be650c04a3136
Author: Roman Khachatryan
AuthorDate: Wed Jun 3 22:12:15 2020 +0200
[FLINK-17869][tests] Unignore UnalignedCheckpointITCase
---
.../org/apache/flink/test/checkpointing/UnalignedCheckpointITCase.java | 2 --
1 file changed, 2 deletions(-)
diff --git
a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointITCase.java
b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointITCase.java
index 47e7007..6211077 100644
---
a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointITCase.java
+++
b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointITCase.java
@@ -44,7 +44,6 @@ import
org.apache.flink.streaming.api.functions.source.RichParallelSourceFunctio
import org.apache.flink.util.TestLogger;
import org.apache.commons.lang3.ArrayUtils;
-import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ErrorCollector;
@@ -102,7 +101,6 @@ import static org.hamcrest.Matchers.greaterThan;
* The number of successful checkpoints is indeed {@code >=n}.
*
*/
-@Ignore
public class UnalignedCheckpointITCase extends TestLogger {
public static final String NUM_INPUTS = "inputs";
public static final String NUM_OUTPUTS = "outputs";
[flink] 04/08: [FLINK-17869][task][checkpointing] Abort channel state write if checkpoint is subsumed
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d8069249703bbe7858e0c6a044deb54ce75e3989
Author: Roman Khachatryan
AuthorDate: Wed Jun 3 14:25:01 2020 +0200
[FLINK-17869][task][checkpointing] Abort channel state write if checkpoint
is subsumed
Motivation: stop writing channel state ASAP if the checkpoint is subsumed
Changes:
1. complete
CheckpointBarrierUnaligner.ThreadSafeUnaligner#allBarriersReceivedFuture
2. abort channel state write on its erroneous completion
3. add cleanup parameter to ChannelStateWriter.abort to use cleanup=false
in the call above
---
.../checkpoint/channel/ChannelStateWriter.java | 4 +--
.../checkpoint/channel/ChannelStateWriterImpl.java | 6 +++--
.../channel/ChannelStateWriterImplTest.java| 2 +-
.../checkpoint/channel/MockChannelStateWriter.java | 2 +-
.../runtime/io/CheckpointBarrierUnaligner.java | 31 +++---
.../tasks/SubtaskCheckpointCoordinatorImpl.java| 15 +++
.../tasks/TestSubtaskCheckpointCoordinator.java| 2 +-
7 files changed, 35 insertions(+), 27 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
index 02a3a69..2112444 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
@@ -139,7 +139,7 @@ public interface ChannelStateWriter extends Closeable {
* Aborts the checkpoint and fails pending result for this checkpoint.
* @param cleanup true if {@link #getAndRemoveWriteResult(long)} is not
supposed to be called afterwards.
*/
- void abort(long checkpointId, Throwable cause);
+ void abort(long checkpointId, Throwable cause, boolean cleanup);
/**
* Must be called after {@link #start(long, CheckpointOptions)} once.
@@ -174,7 +174,7 @@ public interface ChannelStateWriter extends Closeable {
}
@Override
- public void abort(long checkpointId, Throwable cause) {
+ public void abort(long checkpointId, Throwable cause, boolean
cleanup) {
}
@Override
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
index 8996b3b..3e18050 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
@@ -144,11 +144,13 @@ public class ChannelStateWriterImpl implements
ChannelStateWriter {
}
@Override
- public void abort(long checkpointId, Throwable cause) {
+ public void abort(long checkpointId, Throwable cause, boolean cleanup) {
LOG.debug("{} aborting, checkpoint {}", taskName, checkpointId);
enqueue(ChannelStateWriteRequest.abort(checkpointId, cause),
true); // abort already started
enqueue(ChannelStateWriteRequest.abort(checkpointId, cause),
false); // abort enqueued but not started
- results.remove(checkpointId);
+ if (cleanup) {
+ results.remove(checkpointId);
+ }
}
@Override
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
index 0dae88e..d0193dd 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
@@ -290,7 +290,7 @@ public class ChannelStateWriterImplTest {
}
private void callAbort(ChannelStateWriter writer) {
- writer.abort(CHECKPOINT_ID, new TestException());
+ writer.abort(CHECKPOINT_ID, new TestException(), false);
}
private void callFinish(ChannelStateWriter writer) {
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/MockChannelStateWriter.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/MockChannelStateWriter.java
index 88bd334..7641d36 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/MockChannelStateWriter.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/
[flink] 04/08: [FLINK-17869][task][checkpointing] Abort channel state write if checkpoint is subsumed
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d8069249703bbe7858e0c6a044deb54ce75e3989
Author: Roman Khachatryan
AuthorDate: Wed Jun 3 14:25:01 2020 +0200
[FLINK-17869][task][checkpointing] Abort channel state write if checkpoint
is subsumed
Motivation: stop writing channel state ASAP if the checkpoint is subsumed
Changes:
1. complete
CheckpointBarrierUnaligner.ThreadSafeUnaligner#allBarriersReceivedFuture
2. abort channel state write on its erroneous completion
3. add cleanup parameter to ChannelStateWriter.abort to use cleanup=false
in the call above
---
.../checkpoint/channel/ChannelStateWriter.java | 4 +--
.../checkpoint/channel/ChannelStateWriterImpl.java | 6 +++--
.../channel/ChannelStateWriterImplTest.java| 2 +-
.../checkpoint/channel/MockChannelStateWriter.java | 2 +-
.../runtime/io/CheckpointBarrierUnaligner.java | 31 +++---
.../tasks/SubtaskCheckpointCoordinatorImpl.java| 15 +++
.../tasks/TestSubtaskCheckpointCoordinator.java| 2 +-
7 files changed, 35 insertions(+), 27 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
index 02a3a69..2112444 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
@@ -139,7 +139,7 @@ public interface ChannelStateWriter extends Closeable {
* Aborts the checkpoint and fails pending result for this checkpoint.
* @param cleanup true if {@link #getAndRemoveWriteResult(long)} is not
supposed to be called afterwards.
*/
- void abort(long checkpointId, Throwable cause);
+ void abort(long checkpointId, Throwable cause, boolean cleanup);
/**
* Must be called after {@link #start(long, CheckpointOptions)} once.
@@ -174,7 +174,7 @@ public interface ChannelStateWriter extends Closeable {
}
@Override
- public void abort(long checkpointId, Throwable cause) {
+ public void abort(long checkpointId, Throwable cause, boolean
cleanup) {
}
@Override
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
index 8996b3b..3e18050 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
@@ -144,11 +144,13 @@ public class ChannelStateWriterImpl implements
ChannelStateWriter {
}
@Override
- public void abort(long checkpointId, Throwable cause) {
+ public void abort(long checkpointId, Throwable cause, boolean cleanup) {
LOG.debug("{} aborting, checkpoint {}", taskName, checkpointId);
enqueue(ChannelStateWriteRequest.abort(checkpointId, cause),
true); // abort already started
enqueue(ChannelStateWriteRequest.abort(checkpointId, cause),
false); // abort enqueued but not started
- results.remove(checkpointId);
+ if (cleanup) {
+ results.remove(checkpointId);
+ }
}
@Override
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
index 0dae88e..d0193dd 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
@@ -290,7 +290,7 @@ public class ChannelStateWriterImplTest {
}
private void callAbort(ChannelStateWriter writer) {
- writer.abort(CHECKPOINT_ID, new TestException());
+ writer.abort(CHECKPOINT_ID, new TestException(), false);
}
private void callFinish(ChannelStateWriter writer) {
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/MockChannelStateWriter.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/MockChannelStateWriter.java
index 88bd334..7641d36 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/MockChannelStateWriter.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/
[flink] 01/08: [FLINK-17869][hotfix] Add taskName to ChannelStateWriter log messages
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4030b1b30a2208e494d639af8767c3a944d87d58
Author: Roman Khachatryan
AuthorDate: Wed Jun 3 14:34:30 2020 +0200
[FLINK-17869][hotfix] Add taskName to ChannelStateWriter log messages
---
.../flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java| 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
index e6aa9dc..fc8655c 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
@@ -102,11 +102,11 @@ public class ChannelStateWriterImpl implements
ChannelStateWriter {
LOG.debug("{} starting checkpoint {} ({})", taskName,
checkpointId, checkpointOptions);
ChannelStateWriteResult result = new ChannelStateWriteResult();
ChannelStateWriteResult put =
results.computeIfAbsent(checkpointId, id -> {
- Preconditions.checkState(results.size() <
maxCheckpoints, "results.size() > maxCheckpoints", results.size(),
maxCheckpoints);
+ Preconditions.checkState(results.size() <
maxCheckpoints, String.format("%s can't start %d, results.size() >
maxCheckpoints: %d > %d", taskName, checkpointId, results.size(),
maxCheckpoints));
enqueue(new CheckpointStartRequest(checkpointId,
result, checkpointOptions.getTargetLocation()), false);
return result;
});
- Preconditions.checkArgument(put == result, "result future
already present for checkpoint " + checkpointId);
+ Preconditions.checkArgument(put == result, taskName + " result
future already present for checkpoint " + checkpointId);
}
@Override
@@ -156,7 +156,7 @@ public class ChannelStateWriterImpl implements
ChannelStateWriter {
public ChannelStateWriteResult getWriteResult(long checkpointId) {
LOG.debug("{} requested write result, checkpoint {}", taskName,
checkpointId);
ChannelStateWriteResult result = results.get(checkpointId);
- Preconditions.checkArgument(result != null, "channel state
write result not found for checkpoint " + checkpointId);
+ Preconditions.checkArgument(result != null, taskName + "
channel state write result not found for checkpoint " + checkpointId);
return result;
}
[flink] 07/08: [FLINK-17869][tests] Unignore UnalignedCheckpointITCase
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 0634a2efd2f0ca42d4cb6762015be650c04a3136
Author: Roman Khachatryan
AuthorDate: Wed Jun 3 22:12:15 2020 +0200
[FLINK-17869][tests] Unignore UnalignedCheckpointITCase
---
.../org/apache/flink/test/checkpointing/UnalignedCheckpointITCase.java | 2 --
1 file changed, 2 deletions(-)
diff --git
a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointITCase.java
b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointITCase.java
index 47e7007..6211077 100644
---
a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointITCase.java
+++
b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointITCase.java
@@ -44,7 +44,6 @@ import
org.apache.flink.streaming.api.functions.source.RichParallelSourceFunctio
import org.apache.flink.util.TestLogger;
import org.apache.commons.lang3.ArrayUtils;
-import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ErrorCollector;
@@ -102,7 +101,6 @@ import static org.hamcrest.Matchers.greaterThan;
* The number of successful checkpoints is indeed {@code >=n}.
*
*/
-@Ignore
public class UnalignedCheckpointITCase extends TestLogger {
public static final String NUM_INPUTS = "inputs";
public static final String NUM_OUTPUTS = "outputs";
[flink] 06/08: [FLINK-17869][task][checkpointing] Ignore out of order checkpoints in SubtaskCheckpointCoordinator
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 7bb3ffa91a9916348d2f0a6a2e6cba4b109be56e
Author: Roman Khachatryan
AuthorDate: Wed Jun 3 21:43:56 2020 +0200
[FLINK-17869][task][checkpointing] Ignore out of order checkpoints in
SubtaskCheckpointCoordinator
Check (by task thread) whether the current checkpoint was already aborted
in the following scenario:
1. on checkpoint barrier ThreadSafeUnaligner sends a mail to start
checkpointing (netty thread)
2. on cancellation marker CheckpointBarrierUnaligner aborts it (task thread)
3. task thread processes a mail to start checkpointing
---
.../tasks/SubtaskCheckpointCoordinatorImpl.java| 22 +++---
1 file changed, 19 insertions(+), 3 deletions(-)
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
index cf8a21e..5d7a2c9 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
@@ -44,7 +44,6 @@ import
org.apache.flink.streaming.api.operators.StreamOperator;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.IOUtils;
-import org.apache.flink.util.Preconditions;
import org.apache.flink.util.function.BiFunctionWithException;
import org.slf4j.Logger;
@@ -57,10 +56,12 @@ import java.io.UncheckedIOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
+import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
@@ -183,6 +184,16 @@ class SubtaskCheckpointCoordinatorImpl implements
SubtaskCheckpointCoordinator {
@Override
public void abortCheckpointOnBarrier(long checkpointId, Throwable
cause, OperatorChain operatorChain) throws IOException {
LOG.debug("Aborting checkpoint via cancel-barrier {} for task
{}", checkpointId, taskName);
+ lastCheckpointId = Math.max(lastCheckpointId, checkpointId);
+ Iterator iterator = abortedCheckpointIds.iterator();
+ while (iterator.hasNext()) {
+ long next = iterator.next();
+ if (next < lastCheckpointId) {
+ iterator.remove();
+ } else {
+ break;
+ }
+ }
checkpointStorage.clearCacheFor(checkpointId);
@@ -221,9 +232,14 @@ class SubtaskCheckpointCoordinatorImpl implements
SubtaskCheckpointCoordinator {
// We generally try to emit the checkpoint barrier as soon as
possible to not affect downstream
// checkpoint alignments
+ if (lastCheckpointId >= metadata.getCheckpointId()) {
+ LOG.info("Out of order checkpoint barrier (aborted
previously?): {} >= {}", lastCheckpointId, metadata.getCheckpointId());
+ channelStateWriter.abort(metadata.getCheckpointId(),
new CancellationException(), true);
+ checkAndClearAbortedStatus(metadata.getCheckpointId());
+ return;
+ }
+
// Step (0): Record the last triggered checkpointId.
- Preconditions.checkArgument(lastCheckpointId <
metadata.getCheckpointId(), String.format(
- "Unexpected current checkpoint-id: %s vs last
checkpoint-id: %s", metadata.getCheckpointId(), lastCheckpointId));
lastCheckpointId = metadata.getCheckpointId();
if (checkAndClearAbortedStatus(metadata.getCheckpointId())) {
LOG.info("Checkpoint {} has been notified as aborted,
would not trigger any checkpoint.", metadata.getCheckpointId());
[flink] 05/08: [FLINK-17869][task][checkpointing] Increase ChannelStateWriterImpl.DEFAULT_MAX_CHECKPOINTS
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e14958dcd09f7c32ba2ba25fc6a48696fe7eeca2
Author: Roman Khachatryan
AuthorDate: Wed Jun 3 15:56:47 2020 +0200
[FLINK-17869][task][checkpointing] Increase
ChannelStateWriterImpl.DEFAULT_MAX_CHECKPOINTS
ChannelStateWriter map is cleaned up by the task thread,
so the check in netty thread should take possible delay
into account.
---
.../apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
index 3e18050..89a5247 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
@@ -56,7 +56,7 @@ import static
org.apache.flink.runtime.checkpoint.channel.ChannelStateWriteReque
public class ChannelStateWriterImpl implements ChannelStateWriter {
private static final Logger LOG =
LoggerFactory.getLogger(ChannelStateWriterImpl.class);
- private static final int DEFAULT_MAX_CHECKPOINTS = 5; // currently,
only single in-flight checkpoint is supported
+ private static final int DEFAULT_MAX_CHECKPOINTS = 1000; // includes
max-concurrent-checkpoints + checkpoints to be aborted (scheduled via mailbox)
private final String taskName;
private final ChannelStateWriteRequestExecutor executor;
[flink] 03/08: [FLINK-17869][task][checkpointing] Revert "[FLINK-17218][checkpointing] Ensuring that ChannelStateWriter aborts previous checkpoints before a new checkpoint is started."
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 6cb8f28cf219f79a36b90a567692fafc5f85a2a2
Author: Roman Khachatryan
AuthorDate: Wed Jun 3 14:12:04 2020 +0200
[FLINK-17869][task][checkpointing] Revert "[FLINK-17218][checkpointing]
Ensuring that ChannelStateWriter aborts previous checkpoints before a new
checkpoint is started."
This reverts commit 24ff415f1b76392f75dea7c3538558d24fcb7058
which introduced a race condition when task thread and netty
thread compete for ChannelStateWriteResult.
Instead, next commits fix it by:
1. Map size validation error will be prevented simply by increasing the
limit
2. When a checkpoint is subsumed, it's write result will be removed from on
future completion
---
.../checkpoint/channel/ChannelStateWriter.java| 4 +++-
.../checkpoint/channel/ChannelStateWriterImpl.java| 1 -
.../channel/ChannelStateWriterImplTest.java | 19 ---
3 files changed, 11 insertions(+), 13 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
index af2a708..02a3a69 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
@@ -179,7 +179,9 @@ public interface ChannelStateWriter extends Closeable {
@Override
public ChannelStateWriteResult getAndRemoveWriteResult(long
checkpointId) {
- return ChannelStateWriteResult.EMPTY;
+ return new ChannelStateWriteResult(
+
CompletableFuture.completedFuture(Collections.emptyList()),
+
CompletableFuture.completedFuture(Collections.emptyList()));
}
@Override
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
index 6158358..8996b3b 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
@@ -98,7 +98,6 @@ public class ChannelStateWriterImpl implements
ChannelStateWriter {
@Override
public void start(long checkpointId, CheckpointOptions
checkpointOptions) {
- results.keySet().forEach(oldCheckpointId ->
abort(oldCheckpointId, new Exception("Starting new checkpoint " +
checkpointId)));
LOG.debug("{} starting checkpoint {} ({})", taskName,
checkpointId, checkpointOptions);
ChannelStateWriteResult result = new ChannelStateWriteResult();
ChannelStateWriteResult put =
results.computeIfAbsent(checkpointId, id -> {
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
index 9d7a7ea..0dae88e 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
@@ -186,19 +186,16 @@ public class ChannelStateWriterImplTest {
unwrappingError(TestException.class, () -> callStart(writer));
}
- @Test
- public void testStartAbortsOldCheckpoints() throws Exception {
- int maxCheckpoints = 10;
- runWithSyncWorker((writer, worker) -> {
- writer.start(0,
CheckpointOptions.forCheckpointWithDefaultLocation());
- ChannelStateWriteResult writeResult =
writer.getWriteResult(0);
- for (int i = 1; i <= maxCheckpoints; i++) {
+ @Test(expected = IllegalStateException.class)
+ public void testLimit() throws IOException {
+ int maxCheckpoints = 3;
+ try (ChannelStateWriterImpl writer = new
ChannelStateWriterImpl(TASK_NAME, getStreamFactoryFactory(), maxCheckpoints)) {
+ writer.open();
+ for (int i = 0; i < maxCheckpoints; i++) {
writer.start(i,
CheckpointOptions.forCheckpointWithDefaultLocation());
- worker.processAllRequests();
- assertTrue(writeResult.isDone());
-
[flink] 08/08: [FLINK-17869][task][checkpointing] Abort writing of channel state by RPC notification
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 1d75ec8efe3c31f2a5e1a4572b5cfb8000ddbf67
Author: Roman Khachatryan
AuthorDate: Thu Jun 4 10:06:15 2020 +0200
[FLINK-17869][task][checkpointing] Abort writing of channel state by RPC
notification
---
.../flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java | 2 ++
1 file changed, 2 insertions(+)
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
index 5d7a2c9..d904c87 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
@@ -306,6 +306,8 @@ class SubtaskCheckpointCoordinatorImpl implements
SubtaskCheckpointCoordinator {
}
}
+ channelStateWriter.abort(checkpointId, new
CancellationException("checkpoint aborted via notification"), false);
+
for (StreamOperatorWrapper operatorWrapper :
operatorChain.getAllOperators(true)) {
try {
operatorWrapper.getStreamOperator().notifyCheckpointAborted(checkpointId);
[flink] branch release-1.11 updated (31a17cb -> 1d75ec8)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch release-1.11 in repository https://gitbox.apache.org/repos/asf/flink.git. from 31a17cb [FLINK-18188][Runtime] Derive JM Off-Heap memory from configured Total Flink Memory minus JVM Heap new 4030b1b [FLINK-17869][hotfix] Add taskName to ChannelStateWriter log messages new 36609bb [FLINK-17869][hotfix] Don't pass ChannelStateWrite Future to AsyncCheckpointRunnable new 6cb8f28 [FLINK-17869][task][checkpointing] Revert "[FLINK-17218][checkpointing] Ensuring that ChannelStateWriter aborts previous checkpoints before a new checkpoint is started." new d806924 [FLINK-17869][task][checkpointing] Abort channel state write if checkpoint is subsumed new e14958d [FLINK-17869][task][checkpointing] Increase ChannelStateWriterImpl.DEFAULT_MAX_CHECKPOINTS new 7bb3ffa [FLINK-17869][task][checkpointing] Ignore out of order checkpoints in SubtaskCheckpointCoordinator new 0634a2e [FLINK-17869][tests] Unignore UnalignedCheckpointITCase new 1d75ec8 [FLINK-17869][task][checkpointing] Abort writing of channel state by RPC notification The 8 revisions listed above as "new" are entirely new to this repository and will be described in separate emails. The revisions listed as "add" were already present in the repository and have only been added to this reference. Summary of changes: .../checkpoint/channel/ChannelStateWriter.java | 29 +--- .../checkpoint/channel/ChannelStateWriterImpl.java | 25 -- .../channel/ChannelStateWriterImplTest.java| 49 +--- .../checkpoint/channel/MockChannelStateWriter.java | 8 +--- .../channel/RecordingChannelStateWriter.java | 5 -- .../runtime/state/ChannelPersistenceITCase.java| 2 +- .../runtime/io/CheckpointBarrierUnaligner.java | 31 +++-- .../runtime/tasks/AsyncCheckpointRunnable.java | 6 --- .../tasks/SubtaskCheckpointCoordinatorImpl.java| 54 +- .../runtime/tasks/LocalStateForwardingTest.java| 2 - .../tasks/TestSubtaskCheckpointCoordinator.java| 2 +- .../checkpointing/UnalignedCheckpointITCase.java | 2 - 12 files changed, 95 insertions(+), 120 deletions(-)
[flink] 02/08: [FLINK-17869][hotfix] Don't pass ChannelStateWrite Future to AsyncCheckpointRunnable
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 36609bb5dc2aee4061a47f7a767630f1f5912d96
Author: Roman Khachatryan
AuthorDate: Wed Jun 3 23:03:52 2020 +0200
[FLINK-17869][hotfix] Don't pass ChannelStateWrite Future to
AsyncCheckpointRunnable
OperatorSnapshotFinalizer already waits and holds this future.
ChannelStateWriter.getWriteResult() can then be non-idempotent.
ChannelStateWriter.stop() can then be removed.
---
.../checkpoint/channel/ChannelStateWriter.java | 21 ++--
.../checkpoint/channel/ChannelStateWriterImpl.java | 10 ++--
.../channel/ChannelStateWriterImplTest.java| 28 ++
.../checkpoint/channel/MockChannelStateWriter.java | 6 +
.../channel/RecordingChannelStateWriter.java | 5
.../runtime/state/ChannelPersistenceITCase.java| 2 +-
.../runtime/tasks/AsyncCheckpointRunnable.java | 6 -
.../tasks/SubtaskCheckpointCoordinatorImpl.java| 15 +---
.../runtime/tasks/LocalStateForwardingTest.java| 2 --
9 files changed, 24 insertions(+), 71 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
index 5dad559..af2a708 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
@@ -123,7 +123,7 @@ public interface ChannelStateWriter extends Closeable {
* Finalize write of channel state data for the given checkpoint id.
* Must be called after {@link #start(long, CheckpointOptions)} and all
of the input data of the given checkpoint added.
* When both {@link #finishInput} and {@link #finishOutput} were called
the results can be (eventually) obtained
-* using {@link #getWriteResult}
+* using {@link #getAndRemoveWriteResult}
*/
void finishInput(long checkpointId);
@@ -131,24 +131,21 @@ public interface ChannelStateWriter extends Closeable {
* Finalize write of channel state data for the given checkpoint id.
* Must be called after {@link #start(long, CheckpointOptions)} and all
of the output data of the given checkpoint added.
* When both {@link #finishInput} and {@link #finishOutput} were called
the results can be (eventually) obtained
-* using {@link #getWriteResult}
+* using {@link #getAndRemoveWriteResult}
*/
void finishOutput(long checkpointId);
/**
* Aborts the checkpoint and fails pending result for this checkpoint.
+* @param cleanup true if {@link #getAndRemoveWriteResult(long)} is not
supposed to be called afterwards.
*/
void abort(long checkpointId, Throwable cause);
/**
-* Must be called after {@link #start(long, CheckpointOptions)}.
+* Must be called after {@link #start(long, CheckpointOptions)} once.
+* @throws IllegalArgumentException if the passed checkpointId is not
known.
*/
- ChannelStateWriteResult getWriteResult(long checkpointId);
-
- /**
-* Cleans up the internal state for the given checkpoint.
-*/
- void stop(long checkpointId);
+ ChannelStateWriteResult getAndRemoveWriteResult(long checkpointId)
throws IllegalArgumentException;
ChannelStateWriter NO_OP = new NoOpChannelStateWriter();
@@ -181,16 +178,12 @@ public interface ChannelStateWriter extends Closeable {
}
@Override
- public ChannelStateWriteResult getWriteResult(long
checkpointId) {
+ public ChannelStateWriteResult getAndRemoveWriteResult(long
checkpointId) {
return ChannelStateWriteResult.EMPTY;
}
@Override
public void close() {
}
-
- @Override
- public void stop(long checkpointId) {
- }
}
}
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
index fc8655c..6158358 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
@@ -153,19 +153,13 @@ public class ChannelStateWriterImpl implements
ChannelStateWriter {
}
@Override
- public ChannelStateWriteResult getWriteResult(long checkpointId) {
+ public ChannelStateWriteR
[flink] annotated tag release-1.11.0-rc1 updated (8aaea7f -> 80f862f)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to annotated tag release-1.11.0-rc1 in repository https://gitbox.apache.org/repos/asf/flink.git. *** WARNING: tag release-1.11.0-rc1 was modified! *** from 8aaea7f (commit) to 80f862f (tag) tagging 8aaea7f3486e573f005b64ed0dcedb902139bf12 (commit) replaces pre-apache-rename by Zhijiang on Fri May 22 18:09:55 2020 +0200 - Log - release-1.11.0-rc1 -BEGIN PGP SIGNATURE- iQIzBAABCAAdFiEELahbkyRP36GaYkRQBlPAos6gDQ4FAl7H+VMACgkQBlPAos6g DQ4qSQ//Z6h3qBZcEr7WJj/7s0TK5/jeay7SopEUHFev1X1vKo78D7eV/2u4UqIo fyvjmOtnqHiEIqqXNwl6lKx/uiH8oGP7nlRHvEBEk4NMX5TCuWH3BnWvkDuAgkAc vTjOlF3AjF9E2Fmhj0mchBwggwSau2eRRxDEH3PhUqlLv4HecnchfZPwGT5VfU4Z ykoQVOVU94n77XQPwiikEcSjRBXm1LjE6APnWrGUMpzD20ZmPb9MsHD8hjoWwVKJ 5X0uLjk7Mc3gzHUynFi6PCk9rvZsJMz6yJQWteEacfrlilf1V/YZHZzVRyD8g8v7 9SEwxtX9l/9Bh1yokd+F6zVj8UIUbO/5bt/JkEHa685OUeU8/OKcA2wDduuxf37Y WqK/UrQ/rclRVX+g5AMkI1VoTWtzHeoLf+APjBxyi+yzFmuYbp0Mu2/0sSeSrWH5 58Rwi4zrHSmIO0vCZIX+rvvT6entb4Uz0z/wtYH7kZKmUF/7DCxwN1oZl0v0ZEjK 8tkC6YwfEB9GUHD78EbdsCGWxhyH2MQF6MhyVIAk8tjZUoma72nhZlbR9sNOUTPZ DGoXqOKudw1AmfwR120OIHFH9O3FaM6sGxc5WZknKqL7mSzBJxhYEExL27VrHBYU 3wsJD9RPhKEVLjPYC8J2c5/d2t+2SEWhhPLpLm7V8UzqSeyuKcE= =MdYG -END PGP SIGNATURE- --- No new revisions were added by this update. Summary of changes:
[flink] annotated tag release-1.11.0-rc1 updated (8aaea7f -> 80f862f)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to annotated tag release-1.11.0-rc1 in repository https://gitbox.apache.org/repos/asf/flink.git. *** WARNING: tag release-1.11.0-rc1 was modified! *** from 8aaea7f (commit) to 80f862f (tag) tagging 8aaea7f3486e573f005b64ed0dcedb902139bf12 (commit) replaces pre-apache-rename by Zhijiang on Fri May 22 18:09:55 2020 +0200 - Log - release-1.11.0-rc1 -BEGIN PGP SIGNATURE- iQIzBAABCAAdFiEELahbkyRP36GaYkRQBlPAos6gDQ4FAl7H+VMACgkQBlPAos6g DQ4qSQ//Z6h3qBZcEr7WJj/7s0TK5/jeay7SopEUHFev1X1vKo78D7eV/2u4UqIo fyvjmOtnqHiEIqqXNwl6lKx/uiH8oGP7nlRHvEBEk4NMX5TCuWH3BnWvkDuAgkAc vTjOlF3AjF9E2Fmhj0mchBwggwSau2eRRxDEH3PhUqlLv4HecnchfZPwGT5VfU4Z ykoQVOVU94n77XQPwiikEcSjRBXm1LjE6APnWrGUMpzD20ZmPb9MsHD8hjoWwVKJ 5X0uLjk7Mc3gzHUynFi6PCk9rvZsJMz6yJQWteEacfrlilf1V/YZHZzVRyD8g8v7 9SEwxtX9l/9Bh1yokd+F6zVj8UIUbO/5bt/JkEHa685OUeU8/OKcA2wDduuxf37Y WqK/UrQ/rclRVX+g5AMkI1VoTWtzHeoLf+APjBxyi+yzFmuYbp0Mu2/0sSeSrWH5 58Rwi4zrHSmIO0vCZIX+rvvT6entb4Uz0z/wtYH7kZKmUF/7DCxwN1oZl0v0ZEjK 8tkC6YwfEB9GUHD78EbdsCGWxhyH2MQF6MhyVIAk8tjZUoma72nhZlbR9sNOUTPZ DGoXqOKudw1AmfwR120OIHFH9O3FaM6sGxc5WZknKqL7mSzBJxhYEExL27VrHBYU 3wsJD9RPhKEVLjPYC8J2c5/d2t+2SEWhhPLpLm7V8UzqSeyuKcE= =MdYG -END PGP SIGNATURE- --- No new revisions were added by this update. Summary of changes:
[flink] annotated tag release-1.11.0-rc1 updated (80f862f -> 156ad08b)
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a change to annotated tag release-1.11.0-rc1
in repository https://gitbox.apache.org/repos/asf/flink.git.
*** WARNING: tag release-1.11.0-rc1 was modified! ***
from 80f862f (tag)
to 156ad08b (tag)
tagging 3314b26ea07b97f881a89317b241bbfdeae027ec (commit)
replaces pre-apache-rename
by Zhijiang Wang
on Sun May 24 01:34:58 2020 +0800
- Log -
release-1.11.0-rc1
-BEGIN PGP SIGNATURE-
Version: GnuPG v2.0.22 (GNU/Linux)
iQIcBAABAgAGBQJeyV7CAAoJEAZTwKLOoA0ODNgP/3myMJC+BOiViH/sVGrAzJTi
LfVmLQL/++0qeD2Ug4/jCJ02qtq/eVITCwg7zSSNQCMZzMVPORfpLdeuDWKdAw/7
PA2RzOtS7tiOZftgF58BCcSv51D4Hm28YFpFBYc3NefxCaxmg5LYy55/GP9Ev/jE
89CAhppdNLsCSeybodcYzdefLouPSxR8qlZt77tayJ3O3fQDfe0xBKcX5rn/nGOo
oz4Fd+bHS+ROssoiOkCvu+dGHbpBStgcBRDzTfNQtd8zmOieGUHU1a6KLggsG3nq
Fc2RaiI2IYHQ3k9QI/wMJYAhf/OCIBGbR+psF2D3UjvkZAFMhfcjjTj4njm+jWhu
sO46abagR3+69UliFh5+fnxZmRbvjDHNNk6k7DX421KmKkMJBu+gxw38FxWUQdX6
qyUiMhUTXUFJSkHAz+NU8Rw44udGrBsCZiHdOILPh7zAWxAIMUUYdKh39YC8MB+s
xKJ6o14cBc4lGsLh12G+TFPMV5GphYPLK7P5V05XgFrlq1UspNCnTsWTO0YngQwD
eSr//I2vHYyTV/sBr4D2ejql92xw0bpTKMtman0sWz4wdtjT9RQBW4uFHfKwSeRw
H9TmBgoQKdU+Qaa4VWxeF+epCoT24YC6RJpRFWVj/3qVS6HPq6mEbD7243KRCHYT
S7SXEyVVGtB1qku6EvDL
=hqXK
-END PGP SIGNATURE-
---
discard 8aaea7f Commit for release 1.11.0
add 828ba1d [FLINK-17801][tests] Increase timeout of
TaskExecutorTest.testHeartbeatTimeoutWithResourceManager
add 1bf5b81 [hotfix][tests] Make ClientTest extend TestLogger
add 81c185b [hotfix][qs] Fix logging of exception in RequestWriteListener
add 7cbdd91 [FLINK-13553][qs] Add logging to AbstractServerHandler
add 3e18c10 [FLINK-13553][tests] Enable TRACE logging for
org.apache.flink.queryablestate
add 17d3f05 [FLINK-17854][core] Move InputStatus to flink-core
add a0abaaf [FLINK-17854][core] Let SourceReader use InputStatus directly
add c408bb1 [hotfix][formatting] Fix the checkstyle issue of missing a
javadoc comment in DummyNoOpOperator
new 3314b26 Commit for release 1.11.0
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
annotated tag are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (80f862f)
\
N -- N -- N refs/tags/release-1.11.0-rc1 (156ad08b)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../base/source/reader/SourceReaderBase.java | 13 ++-
.../base/source/reader/SourceReaderTestBase.java | 3 ++-
.../flink/api/connector/source/SourceReader.java | 19 ---
.../org/apache/flink/core}/io/InputStatus.java | 27 +++---
.../connector/source/mocks/MockSourceReader.java | 7 +++---
.../network/AbstractServerHandler.java | 5 +++-
.../flink/queryablestate/network/ClientTest.java | 3 ++-
.../runtime/taskexecutor/TaskExecutorTest.java | 2 +-
.../streaming/api/operators/SourceOperator.java| 13 ++-
.../runtime/io/MultipleInputSelectionHandler.java | 1 +
.../runtime/io/PushingAsyncDataInput.java | 1 +
.../streaming/runtime/io/StreamInputProcessor.java | 1 +
.../runtime/io/StreamMultipleInputProcessor.java | 1 +
.../runtime/io/StreamOneInputProcessor.java| 1 +
.../runtime/io/StreamTaskNetworkInput.java | 1 +
.../runtime/io/StreamTaskSourceInput.java | 1 +
.../runtime/io/StreamTwoInputProcessor.java| 1 +
.../flink/streaming/runtime/tasks/StreamTask.java | 2 +-
.../runtime/io/StreamTaskNetworkInputTest.java | 1 +
.../runtime/tasks/MultipleInputStreamTaskTest.java | 2 +-
.../streaming/runtime/tasks/StreamTaskTest.java| 2 +-
.../apache/flink/table/util/DummyNoOpOperator.java | 3 +++
tools/log4j-travis.properties | 5
23 files changed, 65 insertions(+), 50 deletions(-)
rename {flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime
=> flink-core/src/main/java/org/apache/flink/core}/io/InputStatus.java (52%)
[flink] 01/01: Commit for release 1.11.0
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a commit to annotated tag release-1.11.0-rc1 in repository https://gitbox.apache.org/repos/asf/flink.git commit 3314b26ea07b97f881a89317b241bbfdeae027ec Author: Zhijiang Wang AuthorDate: Sun May 24 01:33:12 2020 +0800 Commit for release 1.11.0 --- docs/_config.yml | 2 +- flink-annotations/pom.xml | 2 +- flink-clients/pom.xml | 2 +- flink-connectors/flink-connector-base/pom.xml | 2 +- flink-connectors/flink-connector-cassandra/pom.xml| 2 +- flink-connectors/flink-connector-elasticsearch-base/pom.xml | 2 +- flink-connectors/flink-connector-elasticsearch5/pom.xml | 2 +- flink-connectors/flink-connector-elasticsearch6/pom.xml | 2 +- flink-connectors/flink-connector-elasticsearch7/pom.xml | 2 +- flink-connectors/flink-connector-filesystem/pom.xml | 2 +- flink-connectors/flink-connector-gcp-pubsub/pom.xml | 2 +- flink-connectors/flink-connector-hbase/pom.xml| 2 +- flink-connectors/flink-connector-hive/pom.xml | 2 +- flink-connectors/flink-connector-jdbc/pom.xml | 2 +- flink-connectors/flink-connector-kafka-0.10/pom.xml | 2 +- flink-connectors/flink-connector-kafka-0.11/pom.xml | 2 +- flink-connectors/flink-connector-kafka-base/pom.xml | 2 +- flink-connectors/flink-connector-kafka/pom.xml| 2 +- flink-connectors/flink-connector-kinesis/pom.xml | 2 +- flink-connectors/flink-connector-nifi/pom.xml | 2 +- flink-connectors/flink-connector-rabbitmq/pom.xml | 2 +- flink-connectors/flink-connector-twitter/pom.xml | 2 +- flink-connectors/flink-hadoop-compatibility/pom.xml | 2 +- flink-connectors/flink-hcatalog/pom.xml | 2 +- flink-connectors/flink-sql-connector-elasticsearch6/pom.xml | 2 +- flink-connectors/flink-sql-connector-elasticsearch7/pom.xml | 2 +- flink-connectors/flink-sql-connector-hive-1.2.2/pom.xml | 2 +- flink-connectors/flink-sql-connector-hive-2.2.0/pom.xml | 2 +- flink-connectors/flink-sql-connector-hive-2.3.6/pom.xml | 2 +- flink-connectors/flink-sql-connector-hive-3.1.2/pom.xml | 2 +- flink-connectors/flink-sql-connector-kafka-0.10/pom.xml | 2 +- flink-connectors/flink-sql-connector-kafka-0.11/pom.xml | 2 +- flink-connectors/flink-sql-connector-kafka/pom.xml| 2 +- flink-connectors/pom.xml | 2 +- flink-container/pom.xml | 2 +- flink-contrib/flink-connector-wikiedits/pom.xml | 2 +- flink-contrib/pom.xml | 2 +- flink-core/pom.xml| 2 +- flink-dist/pom.xml| 2 +- flink-docs/pom.xml| 2 +- flink-end-to-end-tests/flink-batch-sql-test/pom.xml | 2 +- flink-end-to-end-tests/flink-bucketing-sink-test/pom.xml | 2 +- flink-end-to-end-tests/flink-cli-test/pom.xml | 2 +- flink-end-to-end-tests/flink-confluent-schema-registry/pom.xml| 2 +- .../flink-connector-gcp-pubsub-emulator-tests/pom.xml | 2 +- flink-end-to-end-tests/flink-dataset-allround-test/pom.xml| 2 +- .../flink-dataset-fine-grained-recovery-test/pom.xml | 2 +- flink-end-to-end-tests/flink-datastream-allround-test/pom.xml | 2 +- flink-end-to-end-tests/flink-distributed-cache-via-blob-test/pom.xml | 2 +- flink-end-to-end-tests/flink-elasticsearch5-test/pom.xml | 2 +- flink-end-to-end-tests/flink-elasticsearch6-test/pom.xml | 2 +- flink-end-to-end-tests/flink-elasticsearch7-test/pom.xml | 2 +- flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/pom.xml| 2 +- flink-end-to-end-tests/flink-end-to-end-tests-common/pom.xml | 2 +- flink-end-to-end-tests/flink-heavy-deployment-stress-test/pom.xml | 2 +- flink-end-to-end-tests/flink-high-parallelism-iterations-test/pom.xml | 2 +- .../flink-local-recovery-and-allocation-test/pom.xml | 2 +- flink-end-to-end-tests/flink-metrics-availability-test/pom.xml| 2 +- flink-end-to-end-tests/flink-metrics-reporter-prometheus-test/pom.xml | 2 +- .../flink-netty-shuffle-memory-control-test/pom.xml
[flink] annotated tag release-1.11.0-rc1 updated (80f862f -> 156ad08b)
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a change to annotated tag release-1.11.0-rc1
in repository https://gitbox.apache.org/repos/asf/flink.git.
*** WARNING: tag release-1.11.0-rc1 was modified! ***
from 80f862f (tag)
to 156ad08b (tag)
tagging 3314b26ea07b97f881a89317b241bbfdeae027ec (commit)
replaces pre-apache-rename
by Zhijiang Wang
on Sun May 24 01:34:58 2020 +0800
- Log -
release-1.11.0-rc1
-BEGIN PGP SIGNATURE-
Version: GnuPG v2.0.22 (GNU/Linux)
iQIcBAABAgAGBQJeyV7CAAoJEAZTwKLOoA0ODNgP/3myMJC+BOiViH/sVGrAzJTi
LfVmLQL/++0qeD2Ug4/jCJ02qtq/eVITCwg7zSSNQCMZzMVPORfpLdeuDWKdAw/7
PA2RzOtS7tiOZftgF58BCcSv51D4Hm28YFpFBYc3NefxCaxmg5LYy55/GP9Ev/jE
89CAhppdNLsCSeybodcYzdefLouPSxR8qlZt77tayJ3O3fQDfe0xBKcX5rn/nGOo
oz4Fd+bHS+ROssoiOkCvu+dGHbpBStgcBRDzTfNQtd8zmOieGUHU1a6KLggsG3nq
Fc2RaiI2IYHQ3k9QI/wMJYAhf/OCIBGbR+psF2D3UjvkZAFMhfcjjTj4njm+jWhu
sO46abagR3+69UliFh5+fnxZmRbvjDHNNk6k7DX421KmKkMJBu+gxw38FxWUQdX6
qyUiMhUTXUFJSkHAz+NU8Rw44udGrBsCZiHdOILPh7zAWxAIMUUYdKh39YC8MB+s
xKJ6o14cBc4lGsLh12G+TFPMV5GphYPLK7P5V05XgFrlq1UspNCnTsWTO0YngQwD
eSr//I2vHYyTV/sBr4D2ejql92xw0bpTKMtman0sWz4wdtjT9RQBW4uFHfKwSeRw
H9TmBgoQKdU+Qaa4VWxeF+epCoT24YC6RJpRFWVj/3qVS6HPq6mEbD7243KRCHYT
S7SXEyVVGtB1qku6EvDL
=hqXK
-END PGP SIGNATURE-
---
discard 8aaea7f Commit for release 1.11.0
add 828ba1d [FLINK-17801][tests] Increase timeout of
TaskExecutorTest.testHeartbeatTimeoutWithResourceManager
add 1bf5b81 [hotfix][tests] Make ClientTest extend TestLogger
add 81c185b [hotfix][qs] Fix logging of exception in RequestWriteListener
add 7cbdd91 [FLINK-13553][qs] Add logging to AbstractServerHandler
add 3e18c10 [FLINK-13553][tests] Enable TRACE logging for
org.apache.flink.queryablestate
add 17d3f05 [FLINK-17854][core] Move InputStatus to flink-core
add a0abaaf [FLINK-17854][core] Let SourceReader use InputStatus directly
add c408bb1 [hotfix][formatting] Fix the checkstyle issue of missing a
javadoc comment in DummyNoOpOperator
new 3314b26 Commit for release 1.11.0
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
annotated tag are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (80f862f)
\
N -- N -- N refs/tags/release-1.11.0-rc1 (156ad08b)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../base/source/reader/SourceReaderBase.java | 13 ++-
.../base/source/reader/SourceReaderTestBase.java | 3 ++-
.../flink/api/connector/source/SourceReader.java | 19 ---
.../org/apache/flink/core}/io/InputStatus.java | 27 +++---
.../connector/source/mocks/MockSourceReader.java | 7 +++---
.../network/AbstractServerHandler.java | 5 +++-
.../flink/queryablestate/network/ClientTest.java | 3 ++-
.../runtime/taskexecutor/TaskExecutorTest.java | 2 +-
.../streaming/api/operators/SourceOperator.java| 13 ++-
.../runtime/io/MultipleInputSelectionHandler.java | 1 +
.../runtime/io/PushingAsyncDataInput.java | 1 +
.../streaming/runtime/io/StreamInputProcessor.java | 1 +
.../runtime/io/StreamMultipleInputProcessor.java | 1 +
.../runtime/io/StreamOneInputProcessor.java| 1 +
.../runtime/io/StreamTaskNetworkInput.java | 1 +
.../runtime/io/StreamTaskSourceInput.java | 1 +
.../runtime/io/StreamTwoInputProcessor.java| 1 +
.../flink/streaming/runtime/tasks/StreamTask.java | 2 +-
.../runtime/io/StreamTaskNetworkInputTest.java | 1 +
.../runtime/tasks/MultipleInputStreamTaskTest.java | 2 +-
.../streaming/runtime/tasks/StreamTaskTest.java| 2 +-
.../apache/flink/table/util/DummyNoOpOperator.java | 3 +++
tools/log4j-travis.properties | 5
23 files changed, 65 insertions(+), 50 deletions(-)
rename {flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime
=> flink-core/src/main/java/org/apache/flink/core}/io/InputStatus.java (52%)
[flink] 01/01: Commit for release 1.11.0
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a commit to annotated tag release-1.11.0-rc1 in repository https://gitbox.apache.org/repos/asf/flink.git commit 3314b26ea07b97f881a89317b241bbfdeae027ec Author: Zhijiang Wang AuthorDate: Sun May 24 01:33:12 2020 +0800 Commit for release 1.11.0 --- docs/_config.yml | 2 +- flink-annotations/pom.xml | 2 +- flink-clients/pom.xml | 2 +- flink-connectors/flink-connector-base/pom.xml | 2 +- flink-connectors/flink-connector-cassandra/pom.xml| 2 +- flink-connectors/flink-connector-elasticsearch-base/pom.xml | 2 +- flink-connectors/flink-connector-elasticsearch5/pom.xml | 2 +- flink-connectors/flink-connector-elasticsearch6/pom.xml | 2 +- flink-connectors/flink-connector-elasticsearch7/pom.xml | 2 +- flink-connectors/flink-connector-filesystem/pom.xml | 2 +- flink-connectors/flink-connector-gcp-pubsub/pom.xml | 2 +- flink-connectors/flink-connector-hbase/pom.xml| 2 +- flink-connectors/flink-connector-hive/pom.xml | 2 +- flink-connectors/flink-connector-jdbc/pom.xml | 2 +- flink-connectors/flink-connector-kafka-0.10/pom.xml | 2 +- flink-connectors/flink-connector-kafka-0.11/pom.xml | 2 +- flink-connectors/flink-connector-kafka-base/pom.xml | 2 +- flink-connectors/flink-connector-kafka/pom.xml| 2 +- flink-connectors/flink-connector-kinesis/pom.xml | 2 +- flink-connectors/flink-connector-nifi/pom.xml | 2 +- flink-connectors/flink-connector-rabbitmq/pom.xml | 2 +- flink-connectors/flink-connector-twitter/pom.xml | 2 +- flink-connectors/flink-hadoop-compatibility/pom.xml | 2 +- flink-connectors/flink-hcatalog/pom.xml | 2 +- flink-connectors/flink-sql-connector-elasticsearch6/pom.xml | 2 +- flink-connectors/flink-sql-connector-elasticsearch7/pom.xml | 2 +- flink-connectors/flink-sql-connector-hive-1.2.2/pom.xml | 2 +- flink-connectors/flink-sql-connector-hive-2.2.0/pom.xml | 2 +- flink-connectors/flink-sql-connector-hive-2.3.6/pom.xml | 2 +- flink-connectors/flink-sql-connector-hive-3.1.2/pom.xml | 2 +- flink-connectors/flink-sql-connector-kafka-0.10/pom.xml | 2 +- flink-connectors/flink-sql-connector-kafka-0.11/pom.xml | 2 +- flink-connectors/flink-sql-connector-kafka/pom.xml| 2 +- flink-connectors/pom.xml | 2 +- flink-container/pom.xml | 2 +- flink-contrib/flink-connector-wikiedits/pom.xml | 2 +- flink-contrib/pom.xml | 2 +- flink-core/pom.xml| 2 +- flink-dist/pom.xml| 2 +- flink-docs/pom.xml| 2 +- flink-end-to-end-tests/flink-batch-sql-test/pom.xml | 2 +- flink-end-to-end-tests/flink-bucketing-sink-test/pom.xml | 2 +- flink-end-to-end-tests/flink-cli-test/pom.xml | 2 +- flink-end-to-end-tests/flink-confluent-schema-registry/pom.xml| 2 +- .../flink-connector-gcp-pubsub-emulator-tests/pom.xml | 2 +- flink-end-to-end-tests/flink-dataset-allround-test/pom.xml| 2 +- .../flink-dataset-fine-grained-recovery-test/pom.xml | 2 +- flink-end-to-end-tests/flink-datastream-allround-test/pom.xml | 2 +- flink-end-to-end-tests/flink-distributed-cache-via-blob-test/pom.xml | 2 +- flink-end-to-end-tests/flink-elasticsearch5-test/pom.xml | 2 +- flink-end-to-end-tests/flink-elasticsearch6-test/pom.xml | 2 +- flink-end-to-end-tests/flink-elasticsearch7-test/pom.xml | 2 +- flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/pom.xml| 2 +- flink-end-to-end-tests/flink-end-to-end-tests-common/pom.xml | 2 +- flink-end-to-end-tests/flink-heavy-deployment-stress-test/pom.xml | 2 +- flink-end-to-end-tests/flink-high-parallelism-iterations-test/pom.xml | 2 +- .../flink-local-recovery-and-allocation-test/pom.xml | 2 +- flink-end-to-end-tests/flink-metrics-availability-test/pom.xml| 2 +- flink-end-to-end-tests/flink-metrics-reporter-prometheus-test/pom.xml | 2 +- .../flink-netty-shuffle-memory-control-test/pom.xml
[flink] branch master updated (6744033 -> f88d98d)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git. from 6744033 [FLINK-17788][scala-shell] Fix yarn session support in scala shell add f88d98d [FLINK-17182][network][tests] Fix the unstable RemoteInputChannelTest.testConcurrentOnSenderBacklogAndRecycle No new revisions were added by this update. Summary of changes: .../partition/consumer/RemoteInputChannelTest.java | 63 +++--- 1 file changed, 32 insertions(+), 31 deletions(-)
[flink] branch release-1.11 updated: [FLINK-17182][network][tests] Fix the unstable RemoteInputChannelTest.testConcurrentOnSenderBacklogAndRecycle
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.11 by this push:
new 22098b2 [FLINK-17182][network][tests] Fix the unstable
RemoteInputChannelTest.testConcurrentOnSenderBacklogAndRecycle
22098b2 is described below
commit 22098b27c32342f3ef74848a86d4b4d8d3c1dfb8
Author: Yun Gao
AuthorDate: Fri Jun 12 10:52:08 2020 +0800
[FLINK-17182][network][tests] Fix the unstable
RemoteInputChannelTest.testConcurrentOnSenderBacklogAndRecycle
In this unstable unit test, the exclusive buffers and floating buffers are
recycled by different
threads, which might cause unexpected race condition issue. But actually
they should always be
recycled by the same task thread in practice. So we refactor the test
process to recycle them in
the same thread to avoid potential unnecessary issues.
This closes #11924.
---
.../partition/consumer/RemoteInputChannelTest.java | 63 +++---
1 file changed, 32 insertions(+), 31 deletions(-)
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 b280422..3984dde 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
@@ -59,6 +59,8 @@ import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
+import java.util.Queue;
+import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
@@ -862,8 +864,7 @@ public class RemoteInputChannelTest {
// Submit tasks and wait to finish
submitTasksAndWaitForResults(executor, new Callable[]{
- recycleExclusiveBufferTask(inputChannel,
numExclusiveSegments),
- recycleFloatingBufferTask(bufferPool,
numFloatingBuffers),
+ recycleBufferTask(inputChannel, bufferPool,
numExclusiveSegments, numFloatingBuffers),
requestBufferTask});
assertEquals("There should be " +
inputChannel.getNumberOfRequiredBuffers() + " buffers available in channel.",
@@ -912,8 +913,7 @@ public class RemoteInputChannelTest {
// Submit tasks and wait to finish
submitTasksAndWaitForResults(executor, new Callable[]{
- recycleExclusiveBufferTask(inputChannel,
numExclusiveSegments),
- recycleFloatingBufferTask(bufferPool,
numFloatingBuffers),
+ recycleBufferTask(inputChannel, bufferPool,
numExclusiveSegments, numFloatingBuffers),
releaseTask});
assertEquals("There should be no buffers available in
the channel.",
@@ -1222,14 +1222,21 @@ public class RemoteInputChannelTest {
}
/**
-* Requests the exclusive buffers from input channel first and then
recycles them by a callable task.
+* Requests the buffers from input channel and buffer pool first and
then recycles them by a callable task.
*
* @param inputChannel The input channel that exclusive buffers request
from.
+* @param bufferPool The buffer pool that floating buffers request from.
* @param numExclusiveSegments The number of exclusive buffers to
request.
-* @return The callable task to recycle exclusive buffers.
+* @param numFloatingBuffers The number of floating buffers to request.
+* @return The callable task to recycle exclusive and floating buffers.
*/
- private Callable recycleExclusiveBufferTask(RemoteInputChannel
inputChannel, int numExclusiveSegments) {
- final List exclusiveBuffers = new
ArrayList<>(numExclusiveSegments);
+ private Callable recycleBufferTask(
+ RemoteInputChannel inputChannel,
+ BufferPool bufferPool,
+ int numExclusiveSegments,
+ int numFloatingBuffers) throws Exception {
+
+ Queue exclusiveBuffers = new
ArrayDeque<>(numExclusiveSegments);
// Exhaust all the exclusive buffers
for (int i = 0; i < numExclusiveSegments; i++) {
Buffer buffer = inputChannel.requestBuffer();
@@ -1237,27 +1244,7 @@ public class RemoteInputChannelTest {
[flink] branch master updated: [FLINK-18252][checkpointing] Fix savepoint overtaking output data.
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new dd3717e [FLINK-18252][checkpointing] Fix savepoint overtaking output
data.
dd3717e is described below
commit dd3717e8383c39eb086f21fba79db0381dfac248
Author: Arvid Heise
AuthorDate: Thu Jun 11 21:18:11 2020 +0200
[FLINK-18252][checkpointing] Fix savepoint overtaking output data.
Currently, a checkpoint/savepoint barrier is always send as a priority
events to the output partitions, where it overtakes data. After the fix a
barrier is only a priority event iff it's unaligned.
Also CheckpointCoordinator only set unaligned flag if the barrier belongs
to a checkpoint.
Ultimately, the unaligned checkpoint config option is not used by
SubtaskCheckpointCoordinatorImpl except for initializing the channel state
writer. The source of truth is now the CheckpointOptions.
---
.../runtime/checkpoint/CheckpointCoordinator.java | 2 +-
.../tasks/SubtaskCheckpointCoordinatorImpl.java| 16 +++--
.../tasks/SubtaskCheckpointCoordinatorTest.java| 40 +++---
3 files changed, 41 insertions(+), 17 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
index da518ee..735c343 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
@@ -732,7 +732,7 @@ public class CheckpointCoordinator {
props.getCheckpointType(),
checkpointStorageLocation.getLocationReference(),
isExactlyOnceMode,
- unalignedCheckpointsEnabled);
+ props.getCheckpointType() == CheckpointType.CHECKPOINT
&& unalignedCheckpointsEnabled);
// send the messages to the tasks that trigger their checkpoint
for (Execution execution: executions) {
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
index d904c87..b2fe147 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
@@ -82,7 +82,6 @@ class SubtaskCheckpointCoordinatorImpl implements
SubtaskCheckpointCoordinator {
private final AsyncExceptionHandler asyncExceptionHandler;
private final ChannelStateWriter channelStateWriter;
private final StreamTaskActionExecutor actionExecutor;
- private final boolean unalignedCheckpointEnabled;
private final BiFunctionWithException, IOException> prepareInputSnapshot;
/** The IDs of the checkpoint for which we are notified aborted. */
private final Set abortedCheckpointIds;
@@ -139,7 +138,6 @@ class SubtaskCheckpointCoordinatorImpl implements
SubtaskCheckpointCoordinator {
executorService,
env,
asyncExceptionHandler,
- unalignedCheckpointEnabled,
prepareInputSnapshot,
maxRecordAbortedCheckpoints,
unalignedCheckpointEnabled ?
openChannelStateWriter(taskName, checkpointStorage) : ChannelStateWriter.NO_OP);
@@ -154,7 +152,6 @@ class SubtaskCheckpointCoordinatorImpl implements
SubtaskCheckpointCoordinator {
ExecutorService executorService,
Environment env,
AsyncExceptionHandler asyncExceptionHandler,
- boolean unalignedCheckpointEnabled,
BiFunctionWithException, IOException> prepareInputSnapshot,
int maxRecordAbortedCheckpoints,
ChannelStateWriter channelStateWriter) throws
IOException {
@@ -167,7 +164,6 @@ class SubtaskCheckpointCoordinatorImpl implements
SubtaskCheckpointCoordinator {
this.asyncExceptionHandler =
checkNotNull(asyncExceptionHandler);
this.actionExecutor = checkNotNull(actionExecutor);
this.channelStateWriter = checkNotNull(channelStateWriter);
- this.unalignedCheckpointEnabled = unalignedCheckpointEnabled;
this.prepareInputSnapshot = prepareInputSnapshot;
this.abortedCheckpointIds =
createAbortedCheckpointSetWithLimitSize(maxRecor
[flink] branch master updated (4d8f55e -> 7e7b132)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git. from 4d8f55e [FLINK-18259][tests] Increase heartbeat timeouts for HeartbeatManagerTest add ca9dc71 [FLINK-17322][network] Fixes BroadcastRecordWriter overwriting memory segments on first finished BufferConsumer. add 7e7b132 [FLINK-17322][network] Disallowing repeated consumer creation for BufferBuilder. No new revisions were added by this update. Summary of changes: .../network/api/writer/BroadcastRecordWriter.java | 6 +- .../io/network/api/writer/RecordWriter.java| 5 ++ .../runtime/io/network/buffer/BufferBuilder.java | 10 ++- .../runtime/io/network/buffer/BufferConsumer.java | 14 .../api/writer/BroadcastRecordWriterTest.java | 54 + .../api/writer/RecordWriterDelegateTest.java | 10 ++- .../io/network/api/writer/RecordWriterTest.java| 13 ++- .../buffer/BufferBuilderAndConsumerTest.java | 16 +--- .../streaming/runtime/LatencyMarkerITCase.java | 93 ++ 9 files changed, 199 insertions(+), 22 deletions(-) create mode 100644 flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/LatencyMarkerITCase.java
[flink] branch release-1.11 updated (89c224e -> dfdfdad)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch release-1.11 in repository https://gitbox.apache.org/repos/asf/flink.git. from 89c224e [FLINK-18265][fs-connector] temp path in FileSystemOutputFormat should be deleted new aaa3bbc [FLINK-17322][network] Fixes BroadcastRecordWriter overwriting memory segments on first finished BufferConsumer. new dfdfdad [FLINK-17322][network] Disallowing repeated consumer creation for BufferBuilder. The 2 revisions listed above as "new" are entirely new to this repository and will be described in separate emails. The revisions listed as "add" were already present in the repository and have only been added to this reference. Summary of changes: .../network/api/writer/BroadcastRecordWriter.java | 6 +- .../io/network/api/writer/RecordWriter.java| 5 ++ .../runtime/io/network/buffer/BufferBuilder.java | 10 ++- .../runtime/io/network/buffer/BufferConsumer.java | 14 .../api/writer/BroadcastRecordWriterTest.java | 54 + .../api/writer/RecordWriterDelegateTest.java | 10 ++- .../io/network/api/writer/RecordWriterTest.java| 13 ++- .../buffer/BufferBuilderAndConsumerTest.java | 16 +--- .../streaming/runtime/LatencyMarkerITCase.java | 93 ++ 9 files changed, 199 insertions(+), 22 deletions(-) create mode 100644 flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/LatencyMarkerITCase.java
[flink] 01/02: [FLINK-17322][network] Fixes BroadcastRecordWriter overwriting memory segments on first finished BufferConsumer.
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit aaa3bbc8495474aea03b2e5184e5a15296ed8809
Author: Arvid Heise
AuthorDate: Wed Jun 3 13:48:49 2020 +0200
[FLINK-17322][network] Fixes BroadcastRecordWriter overwriting memory
segments on first finished BufferConsumer.
BroadcastRecordWriter#randomEmit initialized buffer consumers for other
non-target channels incorrectly leading to separate buffer reference counting
and subsequently released buffers too early.
This commit uses the new BufferConsumer#copyWithReaderPosition method to
copy the buffer while updating the read index to the last committed write index
of the builder.
---
.../network/api/writer/BroadcastRecordWriter.java | 6 +-
.../runtime/io/network/buffer/BufferBuilder.java | 6 +-
.../runtime/io/network/buffer/BufferConsumer.java | 14
.../api/writer/BroadcastRecordWriterTest.java | 54 +
.../io/network/api/writer/RecordWriterTest.java| 8 +-
.../streaming/runtime/LatencyMarkerITCase.java | 93 ++
6 files changed, 176 insertions(+), 5 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/BroadcastRecordWriter.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/BroadcastRecordWriter.java
index 9964b20..132fefa 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/BroadcastRecordWriter.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/BroadcastRecordWriter.java
@@ -50,6 +50,8 @@ public final class BroadcastRecordWriter extends R
*/
private boolean randomTriggered;
+ private BufferConsumer randomTriggeredConsumer;
+
BroadcastRecordWriter(
ResultPartitionWriter writer,
long timeout,
@@ -84,7 +86,7 @@ public final class BroadcastRecordWriter extends R
if (bufferBuilder != null) {
for (int index = 0; index < numberOfChannels; index++) {
if (index != targetChannelIndex) {
-
addBufferConsumer(bufferBuilder.createBufferConsumer(), index);
+
addBufferConsumer(randomTriggeredConsumer.copyWithReaderPosition(bufferBuilder.getCommittedBytes()),
index);
}
}
}
@@ -130,7 +132,7 @@ public final class BroadcastRecordWriter extends R
BufferBuilder builder =
super.requestNewBufferBuilder(targetChannel);
if (randomTriggered) {
- addBufferConsumer(builder.createBufferConsumer(),
targetChannel);
+ addBufferConsumer(randomTriggeredConsumer =
builder.createBufferConsumer(), targetChannel);
} else {
try (BufferConsumer bufferConsumer =
builder.createBufferConsumer()) {
for (int channel = 0; channel <
numberOfChannels; channel++) {
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
index b18569e..7780ba8 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
@@ -123,6 +123,10 @@ public class BufferBuilder {
return getMaxCapacity() - positionMarker.getCached();
}
+ public int getCommittedBytes() {
+ return positionMarker.getCached();
+ }
+
public int getMaxCapacity() {
return memorySegment.size();
}
@@ -167,7 +171,7 @@ public class BufferBuilder {
*
* Remember to commit the {@link SettablePositionMarker} to make the
changes visible.
*/
- private static class SettablePositionMarker implements PositionMarker {
+ static class SettablePositionMarker implements PositionMarker {
private volatile int position = 0;
/**
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferConsumer.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferConsumer.java
index 863b231..70c8457 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferConsumer.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferConsumer.java
@@ -25,6 +25,7 @@ import javax.annotation.concurrent.NotThreadSafe;
import java.io.Closeable;
+import static org.apache.flink.util.Preconditions.checkArgument;
import
[flink] 02/02: [FLINK-17322][network] Disallowing repeated consumer creation for BufferBuilder.
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit dfdfdadf445ea055c841c526b1a382424e1e1865
Author: Arvid Heise
AuthorDate: Fri Jun 12 10:03:59 2020 +0200
[FLINK-17322][network] Disallowing repeated consumer creation for
BufferBuilder.
This is a partial revert of FLINK-10995.
---
.../runtime/io/network/api/writer/RecordWriter.java | 5 +
.../flink/runtime/io/network/buffer/BufferBuilder.java | 4
.../io/network/api/writer/RecordWriterDelegateTest.java | 10 --
.../runtime/io/network/api/writer/RecordWriterTest.java | 5 +++--
.../io/network/buffer/BufferBuilderAndConsumerTest.java | 16 +++-
5 files changed, 23 insertions(+), 17 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java
index 7d0f7de..be40d8d 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java
@@ -349,4 +349,9 @@ public abstract class RecordWriter implements Avai
}
}
}
+
+ @VisibleForTesting
+ ResultPartitionWriter getTargetPartition() {
+ return targetPartition;
+ }
}
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
index 7780ba8..2cb873c 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
@@ -41,6 +41,8 @@ public class BufferBuilder {
private final SettablePositionMarker positionMarker = new
SettablePositionMarker();
+ private boolean bufferConsumerCreated = false;
+
public BufferBuilder(MemorySegment memorySegment, BufferRecycler
recycler) {
this.memorySegment = checkNotNull(memorySegment);
this.recycler = checkNotNull(recycler);
@@ -53,6 +55,8 @@ public class BufferBuilder {
* @return created matching instance of {@link BufferConsumer} to this
{@link BufferBuilder}.
*/
public BufferConsumer createBufferConsumer() {
+ checkState(!bufferConsumerCreated, "Two BufferConsumer
shouldn't exist for one BufferBuilder");
+ bufferConsumerCreated = true;
return new BufferConsumer(
memorySegment,
recycler,
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/writer/RecordWriterDelegateTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/writer/RecordWriterDelegateTest.java
index 97d8053..4a7e5c5 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/writer/RecordWriterDelegateTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/writer/RecordWriterDelegateTest.java
@@ -25,7 +25,9 @@ import
org.apache.flink.runtime.io.network.buffer.BufferBuilderTestUtils;
import org.apache.flink.runtime.io.network.buffer.BufferConsumer;
import org.apache.flink.runtime.io.network.buffer.BufferPool;
import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+import
org.apache.flink.runtime.io.network.partition.NoOpBufferAvailablityListener;
import org.apache.flink.runtime.io.network.partition.ResultPartitionBuilder;
+import org.apache.flink.runtime.io.network.partition.ResultSubpartitionView;
import org.apache.flink.runtime.io.network.partition.consumer.BufferOrEvent;
import org.apache.flink.runtime.io.network.util.TestPooledBufferProvider;
import org.apache.flink.util.TestLogger;
@@ -148,13 +150,17 @@ public class RecordWriterDelegateTest extends TestLogger {
assertTrue(writerDelegate.getAvailableFuture().isDone());
// request one buffer from the local pool to make it unavailable
- final BufferBuilder bufferBuilder =
checkNotNull(writerDelegate.getRecordWriter(0).getBufferBuilder(0));
+ RecordWriter recordWriter = writerDelegate.getRecordWriter(0);
+ final BufferBuilder bufferBuilder =
checkNotNull(recordWriter.getBufferBuilder(0));
assertFalse(writerDelegate.isAvailable());
CompletableFuture future = writerDelegate.getAvailableFuture();
assertFalse(future.isDone());
// recycle the buffer to make the local pool available again
- final Buffer buffer =
BufferBuilderTestUtils.buildSingleBuffe
[flink] 01/02: [FLINK-17322][network] Fixes BroadcastRecordWriter overwriting memory segments on first finished BufferConsumer.
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit aaa3bbc8495474aea03b2e5184e5a15296ed8809
Author: Arvid Heise
AuthorDate: Wed Jun 3 13:48:49 2020 +0200
[FLINK-17322][network] Fixes BroadcastRecordWriter overwriting memory
segments on first finished BufferConsumer.
BroadcastRecordWriter#randomEmit initialized buffer consumers for other
non-target channels incorrectly leading to separate buffer reference counting
and subsequently released buffers too early.
This commit uses the new BufferConsumer#copyWithReaderPosition method to
copy the buffer while updating the read index to the last committed write index
of the builder.
---
.../network/api/writer/BroadcastRecordWriter.java | 6 +-
.../runtime/io/network/buffer/BufferBuilder.java | 6 +-
.../runtime/io/network/buffer/BufferConsumer.java | 14
.../api/writer/BroadcastRecordWriterTest.java | 54 +
.../io/network/api/writer/RecordWriterTest.java| 8 +-
.../streaming/runtime/LatencyMarkerITCase.java | 93 ++
6 files changed, 176 insertions(+), 5 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/BroadcastRecordWriter.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/BroadcastRecordWriter.java
index 9964b20..132fefa 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/BroadcastRecordWriter.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/BroadcastRecordWriter.java
@@ -50,6 +50,8 @@ public final class BroadcastRecordWriter extends R
*/
private boolean randomTriggered;
+ private BufferConsumer randomTriggeredConsumer;
+
BroadcastRecordWriter(
ResultPartitionWriter writer,
long timeout,
@@ -84,7 +86,7 @@ public final class BroadcastRecordWriter extends R
if (bufferBuilder != null) {
for (int index = 0; index < numberOfChannels; index++) {
if (index != targetChannelIndex) {
-
addBufferConsumer(bufferBuilder.createBufferConsumer(), index);
+
addBufferConsumer(randomTriggeredConsumer.copyWithReaderPosition(bufferBuilder.getCommittedBytes()),
index);
}
}
}
@@ -130,7 +132,7 @@ public final class BroadcastRecordWriter extends R
BufferBuilder builder =
super.requestNewBufferBuilder(targetChannel);
if (randomTriggered) {
- addBufferConsumer(builder.createBufferConsumer(),
targetChannel);
+ addBufferConsumer(randomTriggeredConsumer =
builder.createBufferConsumer(), targetChannel);
} else {
try (BufferConsumer bufferConsumer =
builder.createBufferConsumer()) {
for (int channel = 0; channel <
numberOfChannels; channel++) {
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
index b18569e..7780ba8 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
@@ -123,6 +123,10 @@ public class BufferBuilder {
return getMaxCapacity() - positionMarker.getCached();
}
+ public int getCommittedBytes() {
+ return positionMarker.getCached();
+ }
+
public int getMaxCapacity() {
return memorySegment.size();
}
@@ -167,7 +171,7 @@ public class BufferBuilder {
*
* Remember to commit the {@link SettablePositionMarker} to make the
changes visible.
*/
- private static class SettablePositionMarker implements PositionMarker {
+ static class SettablePositionMarker implements PositionMarker {
private volatile int position = 0;
/**
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferConsumer.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferConsumer.java
index 863b231..70c8457 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferConsumer.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferConsumer.java
@@ -25,6 +25,7 @@ import javax.annotation.concurrent.NotThreadSafe;
import java.io.Closeable;
+import static org.apache.flink.util.Preconditions.checkArgument;
import
[flink] 02/02: [FLINK-17322][network] Disallowing repeated consumer creation for BufferBuilder.
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit dfdfdadf445ea055c841c526b1a382424e1e1865
Author: Arvid Heise
AuthorDate: Fri Jun 12 10:03:59 2020 +0200
[FLINK-17322][network] Disallowing repeated consumer creation for
BufferBuilder.
This is a partial revert of FLINK-10995.
---
.../runtime/io/network/api/writer/RecordWriter.java | 5 +
.../flink/runtime/io/network/buffer/BufferBuilder.java | 4
.../io/network/api/writer/RecordWriterDelegateTest.java | 10 --
.../runtime/io/network/api/writer/RecordWriterTest.java | 5 +++--
.../io/network/buffer/BufferBuilderAndConsumerTest.java | 16 +++-
5 files changed, 23 insertions(+), 17 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java
index 7d0f7de..be40d8d 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java
@@ -349,4 +349,9 @@ public abstract class RecordWriter implements Avai
}
}
}
+
+ @VisibleForTesting
+ ResultPartitionWriter getTargetPartition() {
+ return targetPartition;
+ }
}
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
index 7780ba8..2cb873c 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
@@ -41,6 +41,8 @@ public class BufferBuilder {
private final SettablePositionMarker positionMarker = new
SettablePositionMarker();
+ private boolean bufferConsumerCreated = false;
+
public BufferBuilder(MemorySegment memorySegment, BufferRecycler
recycler) {
this.memorySegment = checkNotNull(memorySegment);
this.recycler = checkNotNull(recycler);
@@ -53,6 +55,8 @@ public class BufferBuilder {
* @return created matching instance of {@link BufferConsumer} to this
{@link BufferBuilder}.
*/
public BufferConsumer createBufferConsumer() {
+ checkState(!bufferConsumerCreated, "Two BufferConsumer
shouldn't exist for one BufferBuilder");
+ bufferConsumerCreated = true;
return new BufferConsumer(
memorySegment,
recycler,
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/writer/RecordWriterDelegateTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/writer/RecordWriterDelegateTest.java
index 97d8053..4a7e5c5 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/writer/RecordWriterDelegateTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/writer/RecordWriterDelegateTest.java
@@ -25,7 +25,9 @@ import
org.apache.flink.runtime.io.network.buffer.BufferBuilderTestUtils;
import org.apache.flink.runtime.io.network.buffer.BufferConsumer;
import org.apache.flink.runtime.io.network.buffer.BufferPool;
import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+import
org.apache.flink.runtime.io.network.partition.NoOpBufferAvailablityListener;
import org.apache.flink.runtime.io.network.partition.ResultPartitionBuilder;
+import org.apache.flink.runtime.io.network.partition.ResultSubpartitionView;
import org.apache.flink.runtime.io.network.partition.consumer.BufferOrEvent;
import org.apache.flink.runtime.io.network.util.TestPooledBufferProvider;
import org.apache.flink.util.TestLogger;
@@ -148,13 +150,17 @@ public class RecordWriterDelegateTest extends TestLogger {
assertTrue(writerDelegate.getAvailableFuture().isDone());
// request one buffer from the local pool to make it unavailable
- final BufferBuilder bufferBuilder =
checkNotNull(writerDelegate.getRecordWriter(0).getBufferBuilder(0));
+ RecordWriter recordWriter = writerDelegate.getRecordWriter(0);
+ final BufferBuilder bufferBuilder =
checkNotNull(recordWriter.getBufferBuilder(0));
assertFalse(writerDelegate.isAvailable());
CompletableFuture future = writerDelegate.getAvailableFuture();
assertFalse(future.isDone());
// recycle the buffer to make the local pool available again
- final Buffer buffer =
BufferBuilderTestUtils.buildSingleBuffe
[flink] branch release-1.11 updated (89c224e -> dfdfdad)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch release-1.11 in repository https://gitbox.apache.org/repos/asf/flink.git. from 89c224e [FLINK-18265][fs-connector] temp path in FileSystemOutputFormat should be deleted new aaa3bbc [FLINK-17322][network] Fixes BroadcastRecordWriter overwriting memory segments on first finished BufferConsumer. new dfdfdad [FLINK-17322][network] Disallowing repeated consumer creation for BufferBuilder. The 2 revisions listed above as "new" are entirely new to this repository and will be described in separate emails. The revisions listed as "add" were already present in the repository and have only been added to this reference. Summary of changes: .../network/api/writer/BroadcastRecordWriter.java | 6 +- .../io/network/api/writer/RecordWriter.java| 5 ++ .../runtime/io/network/buffer/BufferBuilder.java | 10 ++- .../runtime/io/network/buffer/BufferConsumer.java | 14 .../api/writer/BroadcastRecordWriterTest.java | 54 + .../api/writer/RecordWriterDelegateTest.java | 10 ++- .../io/network/api/writer/RecordWriterTest.java| 13 ++- .../buffer/BufferBuilderAndConsumerTest.java | 16 +--- .../streaming/runtime/LatencyMarkerITCase.java | 93 ++ 9 files changed, 199 insertions(+), 22 deletions(-) create mode 100644 flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/LatencyMarkerITCase.java
[flink] branch master updated (57992c9 -> bb2affb)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git. from 57992c9 [FLINK-17623][elasticsearch] Support user resource cleanup in ElasticsearchSinkFunction add bb2affb [FLINK-18089][network][tests] Config the e2e for netty shuffle memory control into azure pipeline No new revisions were added by this update. Summary of changes: flink-end-to-end-tests/run-nightly-tests.sh | 2 ++ 1 file changed, 2 insertions(+)
[flink] branch master updated: [FLINK-18089][network][tests] Config the e2e for netty shuffle memory control into azure pipeline
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git The following commit(s) were added to refs/heads/master by this push: new bb2affb [FLINK-18089][network][tests] Config the e2e for netty shuffle memory control into azure pipeline bb2affb is described below commit bb2affbcb6ea748d5a8e8a4af188d4b6f21c1466 Author: Yun Gao AuthorDate: Mon Jun 15 12:29:52 2020 +0800 [FLINK-18089][network][tests] Config the e2e for netty shuffle memory control into azure pipeline This closes #12614. --- flink-end-to-end-tests/run-nightly-tests.sh | 2 ++ 1 file changed, 2 insertions(+) diff --git a/flink-end-to-end-tests/run-nightly-tests.sh b/flink-end-to-end-tests/run-nightly-tests.sh index 69a2ec3..21ea06e 100755 --- a/flink-end-to-end-tests/run-nightly-tests.sh +++ b/flink-end-to-end-tests/run-nightly-tests.sh @@ -181,6 +181,8 @@ run_test "Streaming File Sink end-to-end test" "$END_TO_END_DIR/test-scripts/tes run_test "Streaming File Sink s3 end-to-end test" "$END_TO_END_DIR/test-scripts/test_streaming_file_sink.sh s3" "skip_check_exceptions" run_test "Stateful stream job upgrade end-to-end test" "$END_TO_END_DIR/test-scripts/test_stateful_stream_job_upgrade.sh 2 4" +run_test "Netty shuffle direct memory consumption end-to-end test" "$END_TO_END_DIR/test-scripts/test_netty_shuffle_memory_control.sh" + run_test "Elasticsearch (v5.3.3) sink end-to-end test" "$END_TO_END_DIR/test-scripts/test_streaming_elasticsearch.sh 5 https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-5.3.3.tar.gz"; run_test "Elasticsearch (v6.3.1) sink end-to-end test" "$END_TO_END_DIR/test-scripts/test_streaming_elasticsearch.sh 6 https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-6.3.1.tar.gz";
[flink] branch release-1.11 updated: [FLINK-18089][network][tests] Config the e2e for netty shuffle memory control into azure pipeline
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a commit to branch release-1.11 in repository https://gitbox.apache.org/repos/asf/flink.git The following commit(s) were added to refs/heads/release-1.11 by this push: new 75175d2 [FLINK-18089][network][tests] Config the e2e for netty shuffle memory control into azure pipeline 75175d2 is described below commit 75175d2e5a20e09ec31261e9e098b0e1e0d52a16 Author: Yun Gao AuthorDate: Mon Jun 15 13:48:55 2020 +0800 [FLINK-18089][network][tests] Config the e2e for netty shuffle memory control into azure pipeline --- flink-end-to-end-tests/run-nightly-tests.sh | 2 ++ 1 file changed, 2 insertions(+) diff --git a/flink-end-to-end-tests/run-nightly-tests.sh b/flink-end-to-end-tests/run-nightly-tests.sh index 3a3471c..04422c7 100755 --- a/flink-end-to-end-tests/run-nightly-tests.sh +++ b/flink-end-to-end-tests/run-nightly-tests.sh @@ -181,6 +181,8 @@ run_test "Streaming File Sink end-to-end test" "$END_TO_END_DIR/test-scripts/tes run_test "Streaming File Sink s3 end-to-end test" "$END_TO_END_DIR/test-scripts/test_streaming_file_sink.sh s3" "skip_check_exceptions" run_test "Stateful stream job upgrade end-to-end test" "$END_TO_END_DIR/test-scripts/test_stateful_stream_job_upgrade.sh 2 4" +run_test "Netty shuffle direct memory consumption end-to-end test" "$END_TO_END_DIR/test-scripts/test_netty_shuffle_memory_control.sh" + run_test "Elasticsearch (v5.3.3) sink end-to-end test" "$END_TO_END_DIR/test-scripts/test_streaming_elasticsearch.sh 5 https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-5.3.3.tar.gz"; run_test "Elasticsearch (v6.3.1) sink end-to-end test" "$END_TO_END_DIR/test-scripts/test_streaming_elasticsearch.sh 6 https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-6.3.1.tar.gz";
[flink] branch release-1.10 updated (2bf5ea2 -> 5c0f827)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch release-1.10 in repository https://gitbox.apache.org/repos/asf/flink.git. from 2bf5ea2 [FLINK-17891][yarn] Set execution.target=yarn-session in FlinkYarnSessionCli.run() add 5fc4d1b [FLINK-17322][network] Fixes BroadcastRecordWriter overwriting memory segments on first finished BufferConsumer. add 5c0f827 [FLINK-17322][network] Disallowing repeated consumer creation for BufferBuilder. No new revisions were added by this update. Summary of changes: .../network/api/writer/BroadcastRecordWriter.java | 6 +- .../io/network/api/writer/RecordWriter.java| 5 ++ .../runtime/io/network/buffer/BufferBuilder.java | 8 ++ .../runtime/io/network/buffer/BufferConsumer.java | 14 .../api/writer/BroadcastRecordWriterTest.java | 54 + .../api/writer/RecordWriterDelegateTest.java | 12 ++- .../io/network/api/writer/RecordWriterTest.java| 73 - .../buffer/BufferBuilderAndConsumerTest.java | 16 +--- .../partition/NoOpBufferAvailablityListener.java | 2 +- .../streaming/runtime/LatencyMarkerITCase.java | 93 ++ 10 files changed, 264 insertions(+), 19 deletions(-) create mode 100644 flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/LatencyMarkerITCase.java
[flink] branch release-1.10 updated (2bf5ea2 -> 5c0f827)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch release-1.10 in repository https://gitbox.apache.org/repos/asf/flink.git. from 2bf5ea2 [FLINK-17891][yarn] Set execution.target=yarn-session in FlinkYarnSessionCli.run() add 5fc4d1b [FLINK-17322][network] Fixes BroadcastRecordWriter overwriting memory segments on first finished BufferConsumer. add 5c0f827 [FLINK-17322][network] Disallowing repeated consumer creation for BufferBuilder. No new revisions were added by this update. Summary of changes: .../network/api/writer/BroadcastRecordWriter.java | 6 +- .../io/network/api/writer/RecordWriter.java| 5 ++ .../runtime/io/network/buffer/BufferBuilder.java | 8 ++ .../runtime/io/network/buffer/BufferConsumer.java | 14 .../api/writer/BroadcastRecordWriterTest.java | 54 + .../api/writer/RecordWriterDelegateTest.java | 12 ++- .../io/network/api/writer/RecordWriterTest.java| 73 - .../buffer/BufferBuilderAndConsumerTest.java | 16 +--- .../partition/NoOpBufferAvailablityListener.java | 2 +- .../streaming/runtime/LatencyMarkerITCase.java | 93 ++ 10 files changed, 264 insertions(+), 19 deletions(-) create mode 100644 flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/LatencyMarkerITCase.java
[flink] branch master updated: [FLINK-18238][checkpoint] Broadcast CancelCheckpointMarker while executing checkpoint aborted by coordinator RPC
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 07772bd [FLINK-18238][checkpoint] Broadcast CancelCheckpointMarker
while executing checkpoint aborted by coordinator RPC
07772bd is described below
commit 07772bdb9abc0bcd3b3c8869f6abdc8088bd7cea
Author: Yun Tang
AuthorDate: Wed Jun 17 11:10:39 2020 +0800
[FLINK-18238][checkpoint] Broadcast CancelCheckpointMarker while executing
checkpoint aborted by coordinator RPC
In the case of aborting checkpoint RPC from CheckpointCoordinator, it will
prevent executing the
respective checkpoint which was already triggered before. But we also need
to broadcast the
CancelCheckpointMarker before exiting the execution , otherwise the
downstream side would
probably wait for barrier alignment until deadlock.
This closes #12664.
---
.../tasks/SubtaskCheckpointCoordinatorImpl.java| 4 +-
.../tasks/SubtaskCheckpointCoordinatorTest.java| 75 ++
2 files changed, 78 insertions(+), 1 deletion(-)
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
index b2fe147..0d6d638 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
@@ -235,9 +235,11 @@ class SubtaskCheckpointCoordinatorImpl implements
SubtaskCheckpointCoordinator {
return;
}
- // Step (0): Record the last triggered checkpointId.
+ // Step (0): Record the last triggered checkpointId and abort
the sync phase of checkpoint if necessary.
lastCheckpointId = metadata.getCheckpointId();
if (checkAndClearAbortedStatus(metadata.getCheckpointId())) {
+ // broadcast cancel checkpoint marker to avoid
downstream back-pressure due to checkpoint barrier align.
+ operatorChain.broadcastEvent(new
CancelCheckpointMarker(metadata.getCheckpointId()));
LOG.info("Checkpoint {} has been notified as aborted,
would not trigger any checkpoint.", metadata.getCheckpointId());
return;
}
diff --git
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorTest.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorTest.java
index f7655ea..00ef9a1 100644
---
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorTest.java
+++
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorTest.java
@@ -18,6 +18,9 @@
package org.apache.flink.streaming.runtime.tasks;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
import org.apache.flink.core.fs.CloseableRegistry;
import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.metrics.MetricGroup;
@@ -28,7 +31,11 @@ import org.apache.flink.runtime.checkpoint.CheckpointType;
import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriterImpl;
import org.apache.flink.runtime.event.AbstractEvent;
+import org.apache.flink.runtime.io.network.api.CancelCheckpointMarker;
import org.apache.flink.runtime.io.network.api.writer.NonRecordWriter;
+import
org.apache.flink.runtime.io.network.api.writer.RecordOrEventCollectingResultPartitionWriter;
+import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
+import org.apache.flink.runtime.io.network.util.TestPooledBufferProvider;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.operators.testutils.DummyEnvironment;
import org.apache.flink.runtime.operators.testutils.MockEnvironment;
@@ -39,12 +46,15 @@ import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.SnapshotResult;
import org.apache.flink.runtime.state.TestCheckpointStorageWorkerView;
import org.apache.flink.runtime.state.TestTaskStateManager;
+import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.OperatorSnaps
[flink] branch master updated (e2db1dc -> 07772bd)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git. from e2db1dc [FLINK-18303][filesystem][hive] Fix Filesystem connector doesn't flush part files after rolling interval add 07772bd [FLINK-18238][checkpoint] Broadcast CancelCheckpointMarker while executing checkpoint aborted by coordinator RPC No new revisions were added by this update. Summary of changes: .../tasks/SubtaskCheckpointCoordinatorImpl.java| 4 +- .../tasks/SubtaskCheckpointCoordinatorTest.java| 75 ++ 2 files changed, 78 insertions(+), 1 deletion(-)
[flink] branch release-1.11 updated: [FLINK-18238][checkpoint] Broadcast CancelCheckpointMarker while executing checkpoint aborted by coordinator RPC
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.11 by this push:
new e7c634f [FLINK-18238][checkpoint] Broadcast CancelCheckpointMarker
while executing checkpoint aborted by coordinator RPC
e7c634f is described below
commit e7c634f223c0a2c180ca5abe14a4661115f0afe6
Author: Yun Tang
AuthorDate: Wed Jun 17 21:10:02 2020 +0800
[FLINK-18238][checkpoint] Broadcast CancelCheckpointMarker while executing
checkpoint aborted by coordinator RPC
In the case of aborting checkpoint RPC from CheckpointCoordinator, it will
prevent executing the
respective checkpoint which was already triggered before. But we also need
to broadcast the
CancelCheckpointMarker before exiting the execution , otherwise the
downstream side would
probably wait for barrier alignment until deadlock.
This closes #12664.
---
.../tasks/SubtaskCheckpointCoordinatorImpl.java| 4 +-
.../tasks/SubtaskCheckpointCoordinatorTest.java| 75 ++
2 files changed, 78 insertions(+), 1 deletion(-)
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
index b2fe147..0d6d638 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
@@ -235,9 +235,11 @@ class SubtaskCheckpointCoordinatorImpl implements
SubtaskCheckpointCoordinator {
return;
}
- // Step (0): Record the last triggered checkpointId.
+ // Step (0): Record the last triggered checkpointId and abort
the sync phase of checkpoint if necessary.
lastCheckpointId = metadata.getCheckpointId();
if (checkAndClearAbortedStatus(metadata.getCheckpointId())) {
+ // broadcast cancel checkpoint marker to avoid
downstream back-pressure due to checkpoint barrier align.
+ operatorChain.broadcastEvent(new
CancelCheckpointMarker(metadata.getCheckpointId()));
LOG.info("Checkpoint {} has been notified as aborted,
would not trigger any checkpoint.", metadata.getCheckpointId());
return;
}
diff --git
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorTest.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorTest.java
index f7655ea..00ef9a1 100644
---
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorTest.java
+++
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorTest.java
@@ -18,6 +18,9 @@
package org.apache.flink.streaming.runtime.tasks;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
import org.apache.flink.core.fs.CloseableRegistry;
import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.metrics.MetricGroup;
@@ -28,7 +31,11 @@ import org.apache.flink.runtime.checkpoint.CheckpointType;
import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriterImpl;
import org.apache.flink.runtime.event.AbstractEvent;
+import org.apache.flink.runtime.io.network.api.CancelCheckpointMarker;
import org.apache.flink.runtime.io.network.api.writer.NonRecordWriter;
+import
org.apache.flink.runtime.io.network.api.writer.RecordOrEventCollectingResultPartitionWriter;
+import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
+import org.apache.flink.runtime.io.network.util.TestPooledBufferProvider;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.operators.testutils.DummyEnvironment;
import org.apache.flink.runtime.operators.testutils.MockEnvironment;
@@ -39,12 +46,15 @@ import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.SnapshotResult;
import org.apache.flink.runtime.state.TestCheckpointStorageWorkerView;
import org.apache.flink.runtime.state.TestTaskStateManager;
+import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.OneInputStreamOperato
[flink] annotated tag release-1.11.0-rc2 updated (c4132de -> 1934b4f)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to annotated tag release-1.11.0-rc2 in repository https://gitbox.apache.org/repos/asf/flink.git. *** WARNING: tag release-1.11.0-rc2 was modified! *** from c4132de (commit) to 1934b4f (tag) tagging c4132de4a50ab9b8f653c69af1ba15af44ff29a2 (commit) replaces pre-apache-rename by zhijiang on Thu Jun 18 10:35:21 2020 +0800 - Log - release-1.11.0-rc2 -BEGIN PGP SIGNATURE- Version: GnuPG v2.0.22 (GNU/Linux) iQIcBAABAgAGBQJe6tLpAAoJEAZTwKLOoA0OQaMQALAisIDETJbt4/Vgku2psVky Bg8N1N31XTZrlWu9ycCyHpv6wJ+nJZABsgGaste8csSPXTLcdWnd3YB4dl2cQfw8 REF8t8CZ1kFomNMHwD74B+xm8uKA+IaDEror+q9lIP6+S66ZX3tLLir/M0AbjQlB kQ1WFcUh3mk8HYlLEiTpOc+fBcRkHJYaMaXpdjz45rGQdDO3T6E1ZkYoxOyVpf41 Cx0don+iwq7CAnwJvHVofgvHcf+HoZtjg/G0WPzCCeSkhU1wXWpRw6SsH4Aoy5Jx UDIY1bnhXccb8jawGDT8T4fZpJ34SLKj6wzEx+RNf6iKEfrBQ+PwkTvJEzg5OZuu zYBgEOidBwYVitl6j/gIkcOpHpucwhdu29UJ6qgXRqTyFkIsyeRpvJuHo+YmGdvZ PAAVIL+HvxtsUsjOP4kfavOIH6ez/7W21tIk1TxMwKbTZrRl6kDfjmhbJRpMttha 0DohvaBhPH7oKvE/UlwEUQTBePhONt5JdwZ5IrNlN+cnDGWUX8y88wyKY8Hvbnip Za9uOdMcqzAW9lOFEe+hm3YoieSTILgLvMaqmSQkrDB/JVR52xcTGY0FDqHIguzT iZj/yr2jXRC9JRFyWOxYIj0eRuR44FN5nyNjC0BtC8q9DBvGkAQosBESpsDlKUWY 1Q3hgX0Zu1pi4vlcQNOJ =Suyk -END PGP SIGNATURE- --- No new revisions were added by this update. Summary of changes:
[flink] annotated tag release-1.11.0-rc2 updated (c4132de -> 1934b4f)
This is an automated email from the ASF dual-hosted git repository. zhijiang pushed a change to annotated tag release-1.11.0-rc2 in repository https://gitbox.apache.org/repos/asf/flink.git. *** WARNING: tag release-1.11.0-rc2 was modified! *** from c4132de (commit) to 1934b4f (tag) tagging c4132de4a50ab9b8f653c69af1ba15af44ff29a2 (commit) replaces pre-apache-rename by zhijiang on Thu Jun 18 10:35:21 2020 +0800 - Log - release-1.11.0-rc2 -BEGIN PGP SIGNATURE- Version: GnuPG v2.0.22 (GNU/Linux) iQIcBAABAgAGBQJe6tLpAAoJEAZTwKLOoA0OQaMQALAisIDETJbt4/Vgku2psVky Bg8N1N31XTZrlWu9ycCyHpv6wJ+nJZABsgGaste8csSPXTLcdWnd3YB4dl2cQfw8 REF8t8CZ1kFomNMHwD74B+xm8uKA+IaDEror+q9lIP6+S66ZX3tLLir/M0AbjQlB kQ1WFcUh3mk8HYlLEiTpOc+fBcRkHJYaMaXpdjz45rGQdDO3T6E1ZkYoxOyVpf41 Cx0don+iwq7CAnwJvHVofgvHcf+HoZtjg/G0WPzCCeSkhU1wXWpRw6SsH4Aoy5Jx UDIY1bnhXccb8jawGDT8T4fZpJ34SLKj6wzEx+RNf6iKEfrBQ+PwkTvJEzg5OZuu zYBgEOidBwYVitl6j/gIkcOpHpucwhdu29UJ6qgXRqTyFkIsyeRpvJuHo+YmGdvZ PAAVIL+HvxtsUsjOP4kfavOIH6ez/7W21tIk1TxMwKbTZrRl6kDfjmhbJRpMttha 0DohvaBhPH7oKvE/UlwEUQTBePhONt5JdwZ5IrNlN+cnDGWUX8y88wyKY8Hvbnip Za9uOdMcqzAW9lOFEe+hm3YoieSTILgLvMaqmSQkrDB/JVR52xcTGY0FDqHIguzT iZj/yr2jXRC9JRFyWOxYIj0eRuR44FN5nyNjC0BtC8q9DBvGkAQosBESpsDlKUWY 1Q3hgX0Zu1pi4vlcQNOJ =Suyk -END PGP SIGNATURE- --- No new revisions were added by this update. Summary of changes:
