You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/02/20 10:22:51 UTC
[GitHub] [flink] wsry opened a new pull request #11155: [FLINK-14818] Fix
receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
wsry opened a new pull request #11155: [FLINK-14818] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155
## What is the purpose of the change
In network benchmark (for example 1000 channels benchmark with 4 record writers) StreamNetworkBenchmarkEnvironment#createInputGate creates 1000 input gates with 4 input channels each, which doesn't make much sense. It is expected that either having 4 receivers with single input gate with 1000 channels each, or a single receiver with 4 input gates, with 1000 channels each.
## Brief change log
- The receiving InputGate setup logic of StreamNetworkBenchmarkEnvironment is changed
## Verifying this change
This change is already covered by existing tests, such as StreamNetworkThroughputBenchmarkTest.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**)
- The serializers: (yes / **no** / don't know)
- The runtime per-record code paths (performance sensitive): (yes / **no** / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes / **no** / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (yes / **no**)
- If yes, how is the feature documented? (**not applicable** / docs / JavaDocs / not documented)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r382399441
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -272,11 +287,161 @@ private static ShuffleDescriptor createShuffleDescriptor(
ResultPartitionID resultPartitionID,
ResourceID location,
TaskManagerLocation senderLocation,
- int channel) {
+ int connectionIndex) {
final NettyShuffleDescriptorBuilder builder = NettyShuffleDescriptorBuilder.newBuilder()
.setId(resultPartitionID)
.setProducerInfoFromTaskManagerLocation(senderLocation)
- .setConnectionIndex(channel);
+ .setConnectionIndex(connectionIndex);
return localMode ? builder.setProducerLocation(location).buildLocal() : builder.buildRemote();
}
+
+ /**
+ * A {@link SingleInputGateFactory} which replaces the default {@link RemoteInputChannel} and
+ * {@link LocalInputChannel} implementation with costume ones.
+ */
+ private static class TestSingleInputGateFactory extends SingleInputGateFactory {
+
+ private final ResourceID taskExecutorResourceId;
+ private final int partitionRequestInitialBackoff;
+ private final int partitionRequestMaxBackoff;
+ private final ConnectionManager connectionManager;
+ private final ResultPartitionManager partitionManager;
+ private final TaskEventPublisher taskEventPublisher;
+ private final NetworkBufferPool networkBufferPool;
+
+ public TestSingleInputGateFactory(
+ @Nonnull ResourceID taskExecutorResourceId,
Review comment:
nit: I am a bit torn for `@Nonnull` with arguments. I know we did it in some cases before like `SingleInputGateFactory`.
In common sense we still consider them as `Nonnull` w/o this annotation. And we only need to emphasis the `Nullable` case to add the annotation.
In another aspect, it seems not very consistent in this class because other constructors do not have it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:FAILURE URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359 TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:FAILURE URL:https://travis-ci.com/flink-ci/flink/builds/149795742 TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
Hash:2d7b20f97e41151ed741a3f6d93756b776b30dc9 Status:UNKNOWN URL:TBD TriggerType:PUSH TriggerID:2d7b20f97e41151ed741a3f6d93756b776b30dc9
-->
## CI report:
* fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149795742) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359)
* 2d7b20f97e41151ed741a3f6d93756b776b30dc9 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:FAILURE URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359 TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:FAILURE URL:https://travis-ci.com/flink-ci/flink/builds/149795742 TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
Hash:2d7b20f97e41151ed741a3f6d93756b776b30dc9 Status:CANCELED URL:https://travis-ci.com/flink-ci/flink/builds/149835681 TriggerType:PUSH TriggerID:2d7b20f97e41151ed741a3f6d93756b776b30dc9
Hash:cfc691c781ae54d63ecaa7c282e6cb34f73425d8 Status:CANCELED URL:https://travis-ci.com/flink-ci/flink/builds/149842861 TriggerType:PUSH TriggerID:cfc691c781ae54d63ecaa7c282e6cb34f73425d8
Hash:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c Status:SUCCESS URL:https://travis-ci.com/flink-ci/flink/builds/149851288 TriggerType:PUSH TriggerID:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c
Hash:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c Status:SUCCESS URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381 TriggerType:PUSH TriggerID:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c
Hash:090a72bd3101d7111490832526937d3973c75795 Status:FAILURE URL:https://travis-ci.com/flink-ci/flink/builds/149946832 TriggerType:PUSH TriggerID:090a72bd3101d7111490832526937d3973c75795
Hash:8fe5e3d33d372b15e15cc52171ea8c3844c02ef9 Status:PENDING URL:https://travis-ci.com/flink-ci/flink/builds/149949398 TriggerType:PUSH TriggerID:8fe5e3d33d372b15e15cc52171ea8c3844c02ef9
-->
## CI report:
* fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149795742) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359)
* 2d7b20f97e41151ed741a3f6d93756b776b30dc9 Travis: [CANCELED](https://travis-ci.com/flink-ci/flink/builds/149835681)
* cfc691c781ae54d63ecaa7c282e6cb34f73425d8 Travis: [CANCELED](https://travis-ci.com/flink-ci/flink/builds/149842861)
* 21b36d4fadd73af33edb4fabc7da9d93c2f2d55c Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/149851288) Azure: [SUCCESS](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381)
* 090a72bd3101d7111490832526937d3973c75795 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149946832)
* 8fe5e3d33d372b15e15cc52171ea8c3844c02ef9 Travis: [PENDING](https://travis-ci.com/flink-ci/flink/builds/149949398)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r382390727
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -272,11 +287,161 @@ private static ShuffleDescriptor createShuffleDescriptor(
ResultPartitionID resultPartitionID,
ResourceID location,
TaskManagerLocation senderLocation,
- int channel) {
+ int connectionIndex) {
final NettyShuffleDescriptorBuilder builder = NettyShuffleDescriptorBuilder.newBuilder()
.setId(resultPartitionID)
.setProducerInfoFromTaskManagerLocation(senderLocation)
- .setConnectionIndex(channel);
+ .setConnectionIndex(connectionIndex);
return localMode ? builder.setProducerLocation(location).buildLocal() : builder.buildRemote();
}
+
+ /**
+ * A {@link SingleInputGateFactory} which replaces the default {@link RemoteInputChannel} and
+ * {@link LocalInputChannel} implementation with costume ones.
+ */
+ private static class TestSingleInputGateFactory extends SingleInputGateFactory {
+
+ private final ResourceID taskExecutorResourceId;
+ private final int partitionRequestInitialBackoff;
+ private final int partitionRequestMaxBackoff;
+ private final ConnectionManager connectionManager;
+ private final ResultPartitionManager partitionManager;
+ private final TaskEventPublisher taskEventPublisher;
+ private final NetworkBufferPool networkBufferPool;
+
+ public TestSingleInputGateFactory(
+ @Nonnull ResourceID taskExecutorResourceId,
+ @Nonnull NettyShuffleEnvironmentConfiguration networkConfig,
+ @Nonnull ConnectionManager connectionManager,
+ @Nonnull ResultPartitionManager partitionManager,
+ @Nonnull TaskEventPublisher taskEventPublisher,
+ @Nonnull NetworkBufferPool networkBufferPool) {
+ super(
+ taskExecutorResourceId,
+ networkConfig,
+ connectionManager,
+ partitionManager,
+ taskEventPublisher,
+ networkBufferPool);
+ this.networkBufferPool = networkBufferPool;
+ this.taskEventPublisher = taskEventPublisher;
+ this.partitionManager = partitionManager;
+ this.connectionManager = connectionManager;
+ this.partitionRequestMaxBackoff = networkConfig.partitionRequestMaxBackoff();
+ this.partitionRequestInitialBackoff = networkConfig.partitionRequestInitialBackoff();
+ this.taskExecutorResourceId = taskExecutorResourceId;
+ }
+
+ @Override
+ protected InputChannel createKnownInputChannel(
+ SingleInputGate inputGate,
+ int index,
+ NettyShuffleDescriptor inputChannelDescriptor,
+ ChannelStatistics channelStatistics,
+ InputChannelMetrics metrics) {
+ ResultPartitionID partitionId = inputChannelDescriptor.getResultPartitionID();
+ if (inputChannelDescriptor.isLocalTo(taskExecutorResourceId)) {
+ return new TestLocalInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics);
+ } else {
+ return new TestRemoteInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ inputChannelDescriptor.getConnectionId(),
+ connectionManager,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics,
+ networkBufferPool);
+ }
+ }
+ }
+
+ /**
+ * A {@link LocalInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestLocalInputChannel extends LocalInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
+
+ public TestLocalInputChannel(
+ SingleInputGate inputGate,
+ int channelIndex,
+ ResultPartitionID partitionId,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ int initialBackoff,
+ int maxBackoff,
+ InputChannelMetrics metrics) {
+ super(
+ inputGate,
+ channelIndex,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ initialBackoff,
+ maxBackoff,
+ metrics);
+ }
+
+ @Override
+ public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
Review comment:
In addition, we might also need to override the method of `retriggerSubpartitionRequest`, in case this process might be used to cause invalid `subpartitionIndex` future.
Same issue for the below remote channel.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149795742",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149835681",
"triggerID" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"triggerType" : "PUSH"
}, {
"hash" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149842861",
"triggerID" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149851288",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "090a72bd3101d7111490832526937d3973c75795",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149946832",
"triggerID" : "090a72bd3101d7111490832526937d3973c75795",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149949398",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5400",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "07c2811b17d4ee4bb517b5b1a98d7a3bc91f2dea",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "07c2811b17d4ee4bb517b5b1a98d7a3bc91f2dea",
"triggerType" : "PUSH"
}, {
"hash" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8fe5e3d33d372b15e15cc52171ea8c3844c02ef9 Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/149949398) Azure: [SUCCESS](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5400)
* 07c2811b17d4ee4bb517b5b1a98d7a3bc91f2dea UNKNOWN
* 8f263fe38efaf89c0f001d3995d64e7bd46493b1 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-591816599
The failure case in travis is unrelated to this PR, merging
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r383088089
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -272,11 +287,161 @@ private static ShuffleDescriptor createShuffleDescriptor(
ResultPartitionID resultPartitionID,
ResourceID location,
TaskManagerLocation senderLocation,
- int channel) {
+ int connectionIndex) {
final NettyShuffleDescriptorBuilder builder = NettyShuffleDescriptorBuilder.newBuilder()
.setId(resultPartitionID)
.setProducerInfoFromTaskManagerLocation(senderLocation)
- .setConnectionIndex(channel);
+ .setConnectionIndex(connectionIndex);
return localMode ? builder.setProducerLocation(location).buildLocal() : builder.buildRemote();
}
+
+ /**
+ * A {@link SingleInputGateFactory} which replaces the default {@link RemoteInputChannel} and
+ * {@link LocalInputChannel} implementation with costume ones.
+ */
+ private static class TestSingleInputGateFactory extends SingleInputGateFactory {
+
+ private final ResourceID taskExecutorResourceId;
+ private final int partitionRequestInitialBackoff;
+ private final int partitionRequestMaxBackoff;
+ private final ConnectionManager connectionManager;
+ private final ResultPartitionManager partitionManager;
+ private final TaskEventPublisher taskEventPublisher;
+ private final NetworkBufferPool networkBufferPool;
+
+ public TestSingleInputGateFactory(
+ @Nonnull ResourceID taskExecutorResourceId,
+ @Nonnull NettyShuffleEnvironmentConfiguration networkConfig,
+ @Nonnull ConnectionManager connectionManager,
+ @Nonnull ResultPartitionManager partitionManager,
+ @Nonnull TaskEventPublisher taskEventPublisher,
+ @Nonnull NetworkBufferPool networkBufferPool) {
+ super(
+ taskExecutorResourceId,
+ networkConfig,
+ connectionManager,
+ partitionManager,
+ taskEventPublisher,
+ networkBufferPool);
+ this.networkBufferPool = networkBufferPool;
+ this.taskEventPublisher = taskEventPublisher;
+ this.partitionManager = partitionManager;
+ this.connectionManager = connectionManager;
+ this.partitionRequestMaxBackoff = networkConfig.partitionRequestMaxBackoff();
+ this.partitionRequestInitialBackoff = networkConfig.partitionRequestInitialBackoff();
+ this.taskExecutorResourceId = taskExecutorResourceId;
+ }
+
+ @Override
+ protected InputChannel createKnownInputChannel(
+ SingleInputGate inputGate,
+ int index,
+ NettyShuffleDescriptor inputChannelDescriptor,
+ ChannelStatistics channelStatistics,
+ InputChannelMetrics metrics) {
+ ResultPartitionID partitionId = inputChannelDescriptor.getResultPartitionID();
+ if (inputChannelDescriptor.isLocalTo(taskExecutorResourceId)) {
+ return new TestLocalInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics);
+ } else {
+ return new TestRemoteInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ inputChannelDescriptor.getConnectionId(),
+ connectionManager,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics,
+ networkBufferPool);
+ }
+ }
+ }
+
+ /**
+ * A {@link LocalInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestLocalInputChannel extends LocalInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
+
+ public TestLocalInputChannel(
+ SingleInputGate inputGate,
+ int channelIndex,
+ ResultPartitionID partitionId,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ int initialBackoff,
+ int maxBackoff,
+ InputChannelMetrics metrics) {
+ super(
+ inputGate,
+ channelIndex,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ initialBackoff,
+ maxBackoff,
+ metrics);
+ }
+
+ @Override
+ public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ super.requestSubpartition(channelIndex);
+ }
+
+ @Override
+ public ResultPartitionID getPartitionId() {
Review comment:
I have some basic thoughts for the concerns of `ResultPartitionID` AND `subpartitionIndex`.
The baseline is that we can not break the existing rules that `ResultPartitionID` is different for every input channel but `subpartitionIndex` is same for all.
- `ResultPartitionID`: Actually we can refactor the previous codes to introduce this final field in `InputChannel` layer, because it is reasonable to maintain a different value for different channels. Then current `SingleInputGata#setInputChannel(IntermediateResultPartitionID, InputChannel)` can be further refactored into `SingleInputGata#setInputChannel(InputChannel)` instead, because `IntermediateResultPartitionID` can be got directly from `InputChannel`.
To do so, we can avoid the hacky override method `getPartitionId` in `TestRemoteInputChannel` below, and we can pass the respective `ResultPartitionID` while constructing the `InputChannel` instance, which can prevent the inconsistent states between inside and outside. It also solves the previous hacky way `SingleInputGata#setInputChannel` in passing, because the passed `IntermediateResultPartitionID` seems redundant which can be got from `SingleInputGate` directly now.
- `subpartitionIndex`: We can keep it same as now. Although we can also maintain this info in the `InputChannel` layer as did above, then we can also pass the desired subpartition index while constructing the `InputChannel` to further void hack `requestSubpartition` method. But considering it is same for all the channels in normal case, so it might be more reasonable to maintain it in gate layer as now.
I guess the first refactoring work might bring more changes, so I can accept the current hack way in this PR to not block the following work. Next I would create a separate ticket for this refactor work future.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149795742",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149835681",
"triggerID" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"triggerType" : "PUSH"
}, {
"hash" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149842861",
"triggerID" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149851288",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "090a72bd3101d7111490832526937d3973c75795",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149946832",
"triggerID" : "090a72bd3101d7111490832526937d3973c75795",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149949398",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5400",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "07c2811b17d4ee4bb517b5b1a98d7a3bc91f2dea",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "07c2811b17d4ee4bb517b5b1a98d7a3bc91f2dea",
"triggerType" : "PUSH"
}, {
"hash" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"status" : "PENDING",
"url" : "https://travis-ci.com/flink-ci/flink/builds/150266416",
"triggerID" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"triggerType" : "PUSH"
}, {
"hash" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5506",
"triggerID" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 07c2811b17d4ee4bb517b5b1a98d7a3bc91f2dea UNKNOWN
* 8f263fe38efaf89c0f001d3995d64e7bd46493b1 Travis: [PENDING](https://travis-ci.com/flink-ci/flink/builds/150266416) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5506)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r382390260
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -272,11 +287,161 @@ private static ShuffleDescriptor createShuffleDescriptor(
ResultPartitionID resultPartitionID,
ResourceID location,
TaskManagerLocation senderLocation,
- int channel) {
+ int connectionIndex) {
final NettyShuffleDescriptorBuilder builder = NettyShuffleDescriptorBuilder.newBuilder()
.setId(resultPartitionID)
.setProducerInfoFromTaskManagerLocation(senderLocation)
- .setConnectionIndex(channel);
+ .setConnectionIndex(connectionIndex);
return localMode ? builder.setProducerLocation(location).buildLocal() : builder.buildRemote();
}
+
+ /**
+ * A {@link SingleInputGateFactory} which replaces the default {@link RemoteInputChannel} and
+ * {@link LocalInputChannel} implementation with costume ones.
+ */
+ private static class TestSingleInputGateFactory extends SingleInputGateFactory {
+
+ private final ResourceID taskExecutorResourceId;
+ private final int partitionRequestInitialBackoff;
+ private final int partitionRequestMaxBackoff;
+ private final ConnectionManager connectionManager;
+ private final ResultPartitionManager partitionManager;
+ private final TaskEventPublisher taskEventPublisher;
+ private final NetworkBufferPool networkBufferPool;
+
+ public TestSingleInputGateFactory(
+ @Nonnull ResourceID taskExecutorResourceId,
+ @Nonnull NettyShuffleEnvironmentConfiguration networkConfig,
+ @Nonnull ConnectionManager connectionManager,
+ @Nonnull ResultPartitionManager partitionManager,
+ @Nonnull TaskEventPublisher taskEventPublisher,
+ @Nonnull NetworkBufferPool networkBufferPool) {
+ super(
+ taskExecutorResourceId,
+ networkConfig,
+ connectionManager,
+ partitionManager,
+ taskEventPublisher,
+ networkBufferPool);
+ this.networkBufferPool = networkBufferPool;
+ this.taskEventPublisher = taskEventPublisher;
+ this.partitionManager = partitionManager;
+ this.connectionManager = connectionManager;
+ this.partitionRequestMaxBackoff = networkConfig.partitionRequestMaxBackoff();
+ this.partitionRequestInitialBackoff = networkConfig.partitionRequestInitialBackoff();
+ this.taskExecutorResourceId = taskExecutorResourceId;
+ }
+
+ @Override
+ protected InputChannel createKnownInputChannel(
+ SingleInputGate inputGate,
+ int index,
+ NettyShuffleDescriptor inputChannelDescriptor,
+ ChannelStatistics channelStatistics,
+ InputChannelMetrics metrics) {
+ ResultPartitionID partitionId = inputChannelDescriptor.getResultPartitionID();
+ if (inputChannelDescriptor.isLocalTo(taskExecutorResourceId)) {
+ return new TestLocalInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics);
+ } else {
+ return new TestRemoteInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ inputChannelDescriptor.getConnectionId(),
+ connectionManager,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics,
+ networkBufferPool);
+ }
+ }
+ }
+
+ /**
+ * A {@link LocalInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestLocalInputChannel extends LocalInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
+
+ public TestLocalInputChannel(
+ SingleInputGate inputGate,
+ int channelIndex,
+ ResultPartitionID partitionId,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ int initialBackoff,
+ int maxBackoff,
+ InputChannelMetrics metrics) {
+ super(
+ inputGate,
+ channelIndex,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ initialBackoff,
+ maxBackoff,
+ metrics);
+ }
+
+ @Override
+ public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ super.requestSubpartition(channelIndex);
+ }
+
+ @Override
+ public ResultPartitionID getPartitionId() {
Review comment:
Why we need to override this method? I guess that the original `partitionId` while constructing the `SingleInputGate` is valid.
Same issue for the below remote channel.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] wsry commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
wsry commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r382426966
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -272,11 +287,161 @@ private static ShuffleDescriptor createShuffleDescriptor(
ResultPartitionID resultPartitionID,
ResourceID location,
TaskManagerLocation senderLocation,
- int channel) {
+ int connectionIndex) {
final NettyShuffleDescriptorBuilder builder = NettyShuffleDescriptorBuilder.newBuilder()
.setId(resultPartitionID)
.setProducerInfoFromTaskManagerLocation(senderLocation)
- .setConnectionIndex(channel);
+ .setConnectionIndex(connectionIndex);
return localMode ? builder.setProducerLocation(location).buildLocal() : builder.buildRemote();
}
+
+ /**
+ * A {@link SingleInputGateFactory} which replaces the default {@link RemoteInputChannel} and
+ * {@link LocalInputChannel} implementation with costume ones.
+ */
+ private static class TestSingleInputGateFactory extends SingleInputGateFactory {
+
+ private final ResourceID taskExecutorResourceId;
+ private final int partitionRequestInitialBackoff;
+ private final int partitionRequestMaxBackoff;
+ private final ConnectionManager connectionManager;
+ private final ResultPartitionManager partitionManager;
+ private final TaskEventPublisher taskEventPublisher;
+ private final NetworkBufferPool networkBufferPool;
+
+ public TestSingleInputGateFactory(
+ @Nonnull ResourceID taskExecutorResourceId,
+ @Nonnull NettyShuffleEnvironmentConfiguration networkConfig,
+ @Nonnull ConnectionManager connectionManager,
+ @Nonnull ResultPartitionManager partitionManager,
+ @Nonnull TaskEventPublisher taskEventPublisher,
+ @Nonnull NetworkBufferPool networkBufferPool) {
+ super(
+ taskExecutorResourceId,
+ networkConfig,
+ connectionManager,
+ partitionManager,
+ taskEventPublisher,
+ networkBufferPool);
+ this.networkBufferPool = networkBufferPool;
+ this.taskEventPublisher = taskEventPublisher;
+ this.partitionManager = partitionManager;
+ this.connectionManager = connectionManager;
+ this.partitionRequestMaxBackoff = networkConfig.partitionRequestMaxBackoff();
+ this.partitionRequestInitialBackoff = networkConfig.partitionRequestInitialBackoff();
+ this.taskExecutorResourceId = taskExecutorResourceId;
+ }
+
+ @Override
+ protected InputChannel createKnownInputChannel(
+ SingleInputGate inputGate,
+ int index,
+ NettyShuffleDescriptor inputChannelDescriptor,
+ ChannelStatistics channelStatistics,
+ InputChannelMetrics metrics) {
+ ResultPartitionID partitionId = inputChannelDescriptor.getResultPartitionID();
+ if (inputChannelDescriptor.isLocalTo(taskExecutorResourceId)) {
+ return new TestLocalInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics);
+ } else {
+ return new TestRemoteInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ inputChannelDescriptor.getConnectionId(),
+ connectionManager,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics,
+ networkBufferPool);
+ }
+ }
+ }
+
+ /**
+ * A {@link LocalInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestLocalInputChannel extends LocalInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
+
+ public TestLocalInputChannel(
+ SingleInputGate inputGate,
+ int channelIndex,
+ ResultPartitionID partitionId,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ int initialBackoff,
+ int maxBackoff,
+ InputChannelMetrics metrics) {
+ super(
+ inputGate,
+ channelIndex,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ initialBackoff,
+ maxBackoff,
+ metrics);
+ }
+
+ @Override
+ public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
Review comment:
I add a comment to the ```createReceiver``` method.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r382394298
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -238,18 +252,19 @@ private InputGate createInputGate(TaskManagerLocation senderLocation) throws Exc
private InputGateDeploymentDescriptor createInputGateDeploymentDescriptor(
TaskManagerLocation senderLocation,
- int consumedSubpartitionIndex,
+ int gateIndex,
ResourceID localLocation) {
- final ShuffleDescriptor[] channelDescriptors = Arrays.stream(partitionIds)
- .map(partitionId ->
- createShuffleDescriptor(localMode, partitionId, localLocation, senderLocation, consumedSubpartitionIndex))
- .toArray(ShuffleDescriptor[]::new);
+ final ShuffleDescriptor[] channelDescriptors = new ShuffleDescriptor[channels];
+ for (int channelIndex = 0; channelIndex < channels; ++channelIndex) {
+ channelDescriptors[channelIndex] = createShuffleDescriptor(
+ localMode, partitionIds[gateIndex], localLocation, senderLocation, channelIndex);
Review comment:
I am a bit torn for the `connectionIndex` value. If we use `channelIndex` as the value, it would create 1000 connections in theory if there are 1000 subpartitions in writer. I a bit prefer to use `gateIndex` instead to create less connections.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:FAILURE URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359 TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:FAILURE URL:https://travis-ci.com/flink-ci/flink/builds/149795742 TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
Hash:2d7b20f97e41151ed741a3f6d93756b776b30dc9 Status:CANCELED URL:https://travis-ci.com/flink-ci/flink/builds/149835681 TriggerType:PUSH TriggerID:2d7b20f97e41151ed741a3f6d93756b776b30dc9
Hash:cfc691c781ae54d63ecaa7c282e6cb34f73425d8 Status:CANCELED URL:https://travis-ci.com/flink-ci/flink/builds/149842861 TriggerType:PUSH TriggerID:cfc691c781ae54d63ecaa7c282e6cb34f73425d8
Hash:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c Status:PENDING URL:https://travis-ci.com/flink-ci/flink/builds/149851288 TriggerType:PUSH TriggerID:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c
-->
## CI report:
* fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149795742) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359)
* 2d7b20f97e41151ed741a3f6d93756b776b30dc9 Travis: [CANCELED](https://travis-ci.com/flink-ci/flink/builds/149835681)
* cfc691c781ae54d63ecaa7c282e6cb34f73425d8 Travis: [CANCELED](https://travis-ci.com/flink-ci/flink/builds/149842861)
* 21b36d4fadd73af33edb4fabc7da9d93c2f2d55c Travis: [PENDING](https://travis-ci.com/flink-ci/flink/builds/149851288)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot commented on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot commented on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:UNKNOWN URL:TBD TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
-->
## CI report:
* fee93cecf80cb73fcdf082f47fe0b9bd727a3002 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r383094730
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/SingleInputGateBenchmarkFactory.java
##########
@@ -0,0 +1,173 @@
+/*
+ * 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.streaming.runtime.io.benchmark;
+
+import org.apache.flink.core.memory.MemorySegmentProvider;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.io.network.ConnectionID;
+import org.apache.flink.runtime.io.network.ConnectionManager;
+import org.apache.flink.runtime.io.network.TaskEventPublisher;
+import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+import org.apache.flink.runtime.io.network.metrics.InputChannelMetrics;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionManager;
+import org.apache.flink.runtime.io.network.partition.consumer.InputChannel;
+import org.apache.flink.runtime.io.network.partition.consumer.LocalInputChannel;
+import org.apache.flink.runtime.io.network.partition.consumer.RemoteInputChannel;
+import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate;
+import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGateFactory;
+import org.apache.flink.runtime.shuffle.NettyShuffleDescriptor;
+import org.apache.flink.runtime.taskmanager.NettyShuffleEnvironmentConfiguration;
+
+import java.io.IOException;
+
+/**
+ * A benchmark-specific input gate factory which overrides the respective methods of creating
+ * {@link RemoteInputChannel} and {@link LocalInputChannel} for requesting specific subpartitions.
+ */
+public class SingleInputGateBenchmarkFactory extends SingleInputGateFactory {
+
+ public SingleInputGateBenchmarkFactory(
+ ResourceID taskExecutorResourceId,
+ NettyShuffleEnvironmentConfiguration networkConfig,
+ ConnectionManager connectionManager,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ NetworkBufferPool networkBufferPool) {
+ super(
+ taskExecutorResourceId,
+ networkConfig,
+ connectionManager,
+ partitionManager,
+ taskEventPublisher,
+ networkBufferPool);
+ }
+
+ @Override
+ protected InputChannel createKnownInputChannel(
+ SingleInputGate inputGate,
+ int index,
+ NettyShuffleDescriptor inputChannelDescriptor,
+ SingleInputGateFactory.ChannelStatistics channelStatistics,
+ InputChannelMetrics metrics) {
+ ResultPartitionID partitionId = inputChannelDescriptor.getResultPartitionID();
+ if (inputChannelDescriptor.isLocalTo(taskExecutorResourceId)) {
+ return new TestLocalInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics);
+ } else {
+ return new TestRemoteInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ inputChannelDescriptor.getConnectionId(),
+ connectionManager,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics,
+ networkBufferPool);
+ }
+ }
+
+ /**
+ * A {@link LocalInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestLocalInputChannel extends LocalInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
Review comment:
nit : final
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r383088089
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -272,11 +287,161 @@ private static ShuffleDescriptor createShuffleDescriptor(
ResultPartitionID resultPartitionID,
ResourceID location,
TaskManagerLocation senderLocation,
- int channel) {
+ int connectionIndex) {
final NettyShuffleDescriptorBuilder builder = NettyShuffleDescriptorBuilder.newBuilder()
.setId(resultPartitionID)
.setProducerInfoFromTaskManagerLocation(senderLocation)
- .setConnectionIndex(channel);
+ .setConnectionIndex(connectionIndex);
return localMode ? builder.setProducerLocation(location).buildLocal() : builder.buildRemote();
}
+
+ /**
+ * A {@link SingleInputGateFactory} which replaces the default {@link RemoteInputChannel} and
+ * {@link LocalInputChannel} implementation with costume ones.
+ */
+ private static class TestSingleInputGateFactory extends SingleInputGateFactory {
+
+ private final ResourceID taskExecutorResourceId;
+ private final int partitionRequestInitialBackoff;
+ private final int partitionRequestMaxBackoff;
+ private final ConnectionManager connectionManager;
+ private final ResultPartitionManager partitionManager;
+ private final TaskEventPublisher taskEventPublisher;
+ private final NetworkBufferPool networkBufferPool;
+
+ public TestSingleInputGateFactory(
+ @Nonnull ResourceID taskExecutorResourceId,
+ @Nonnull NettyShuffleEnvironmentConfiguration networkConfig,
+ @Nonnull ConnectionManager connectionManager,
+ @Nonnull ResultPartitionManager partitionManager,
+ @Nonnull TaskEventPublisher taskEventPublisher,
+ @Nonnull NetworkBufferPool networkBufferPool) {
+ super(
+ taskExecutorResourceId,
+ networkConfig,
+ connectionManager,
+ partitionManager,
+ taskEventPublisher,
+ networkBufferPool);
+ this.networkBufferPool = networkBufferPool;
+ this.taskEventPublisher = taskEventPublisher;
+ this.partitionManager = partitionManager;
+ this.connectionManager = connectionManager;
+ this.partitionRequestMaxBackoff = networkConfig.partitionRequestMaxBackoff();
+ this.partitionRequestInitialBackoff = networkConfig.partitionRequestInitialBackoff();
+ this.taskExecutorResourceId = taskExecutorResourceId;
+ }
+
+ @Override
+ protected InputChannel createKnownInputChannel(
+ SingleInputGate inputGate,
+ int index,
+ NettyShuffleDescriptor inputChannelDescriptor,
+ ChannelStatistics channelStatistics,
+ InputChannelMetrics metrics) {
+ ResultPartitionID partitionId = inputChannelDescriptor.getResultPartitionID();
+ if (inputChannelDescriptor.isLocalTo(taskExecutorResourceId)) {
+ return new TestLocalInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics);
+ } else {
+ return new TestRemoteInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ inputChannelDescriptor.getConnectionId(),
+ connectionManager,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics,
+ networkBufferPool);
+ }
+ }
+ }
+
+ /**
+ * A {@link LocalInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestLocalInputChannel extends LocalInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
+
+ public TestLocalInputChannel(
+ SingleInputGate inputGate,
+ int channelIndex,
+ ResultPartitionID partitionId,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ int initialBackoff,
+ int maxBackoff,
+ InputChannelMetrics metrics) {
+ super(
+ inputGate,
+ channelIndex,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ initialBackoff,
+ maxBackoff,
+ metrics);
+ }
+
+ @Override
+ public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ super.requestSubpartition(channelIndex);
+ }
+
+ @Override
+ public ResultPartitionID getPartitionId() {
Review comment:
I have some basic thoughts for the concerns of `ResultPartitionID` AND `subpartitionIndex`.
The baseline is that we can not break the existing rules that `ResultPartitionID` is different for every input channel but `subpartitionIndex` is same for all.
- `ResultPartitionID`: Actually we can refactor the previous codes to introduce this final field in `InputChannel` layer, because it is reasonable to maintain a different value for different channels. Then current `SingleInputGata#setInputChannel(IntermediateResultPartitionID, InputChannel)` can be further refactored into `SingleInputGata#setInputChannel(InputChannel)` instead, because `IntermediateResultPartitionID` can be got directly from `InputChannel`.
To do so, we can avoid the hacky override method `getPartitionId` in `TestRemoteInputChannel` below, and we can pass the respective `ResultPartitionID` while constructing the `InputChannel` instance, which can prevent the inconsistent states between inside and outside. It also solves the previous hacky way `SingleInputGata#setInputChannel` in passing, because the passed `IntermediateResultPartitionID` seems redundant which can be got from `SingleInputGate` directly now.
- `subpartitionIndex`: We can keep it same as now. Although we can also maintain this info in the `InputChannel` layer as did above, then we can also pass the desired subpartition index while constructing the `InputChannel` to further void hack `requestSubpartition` method. But considering it is same for all the channels in normal case, so it might be more reasonable to maintain it in gate layer as now.
I guess the first refactoring work might bring more changes, so I can accept the current hack way in this PR to not block the following work. Next I would create a separate ticket for this refactor work future.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-590753152
Also attach the comparison results executed in benchmark machine:
Baseline | |
-- | -- | --
org.apache.flink.benchmark.DataSkewStreamNetworkThroughputBenchmarkExecutor.networkSkewedThroughput | 51977.13475 |
org.apache.flink.benchmark.StreamNetworkBroadcastThroughputBenchmarkExecutor.networkBroadcastThroughput | 1164.463375 |
org.apache.flink.benchmark.StreamNetworkThroughputBenchmarkExecutor.networkThroughput | 69553.22069 | 100,100ms
org.apache.flink.benchmark.StreamNetworkThroughputBenchmarkExecutor.networkThroughput | 17430.11747 | 100,100ms,SSL
org.apache.flink.benchmark.StreamNetworkThroughputBenchmarkExecutor.networkThroughput | 51019.51607 | 1000,1ms
org.apache.flink.benchmark.StreamNetworkThroughputBenchmarkExecutor.networkThroughput | 55423.23278 | 1000,100ms
org.apache.flink.benchmark.StreamNetworkThroughputBenchmarkExecutor.networkThroughput | 14924.65899 | 1000,100ms,SSL
PR change | |
org.apache.flink.benchmark.DataSkewStreamNetworkThroughputBenchmarkExecutor.networkSkewedThroughput | 33215.6767 |
org.apache.flink.benchmark.StreamNetworkBroadcastThroughputBenchmarkExecutor.networkBroadcastThroughput | 1189.239694 |
org.apache.flink.benchmark.StreamNetworkThroughputBenchmarkExecutor.networkThroughput | 72450.69576 | 100,100ms
org.apache.flink.benchmark.StreamNetworkThroughputBenchmarkExecutor.networkThroughput | 17018.46785 | 100,100ms,SSL
org.apache.flink.benchmark.StreamNetworkThroughputBenchmarkExecutor.networkThroughput | 45829.91493 | 1000,1ms
org.apache.flink.benchmark.StreamNetworkThroughputBenchmarkExecutor.networkThroughput | 52943.20546 | 1000,100ms
org.apache.flink.benchmark.StreamNetworkThroughputBenchmarkExecutor.networkThroughput | 13762.96631 | 1000,100ms,SSL
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:FAILURE URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359 TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:FAILURE URL:https://travis-ci.com/flink-ci/flink/builds/149795742 TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
Hash:2d7b20f97e41151ed741a3f6d93756b776b30dc9 Status:CANCELED URL:https://travis-ci.com/flink-ci/flink/builds/149835681 TriggerType:PUSH TriggerID:2d7b20f97e41151ed741a3f6d93756b776b30dc9
Hash:cfc691c781ae54d63ecaa7c282e6cb34f73425d8 Status:CANCELED URL:https://travis-ci.com/flink-ci/flink/builds/149842861 TriggerType:PUSH TriggerID:cfc691c781ae54d63ecaa7c282e6cb34f73425d8
Hash:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c Status:SUCCESS URL:https://travis-ci.com/flink-ci/flink/builds/149851288 TriggerType:PUSH TriggerID:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c
Hash:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c Status:SUCCESS URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381 TriggerType:PUSH TriggerID:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c
Hash:090a72bd3101d7111490832526937d3973c75795 Status:UNKNOWN URL:TBD TriggerType:PUSH TriggerID:090a72bd3101d7111490832526937d3973c75795
-->
## CI report:
* fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149795742) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359)
* 2d7b20f97e41151ed741a3f6d93756b776b30dc9 Travis: [CANCELED](https://travis-ci.com/flink-ci/flink/builds/149835681)
* cfc691c781ae54d63ecaa7c282e6cb34f73425d8 Travis: [CANCELED](https://travis-ci.com/flink-ci/flink/builds/149842861)
* 21b36d4fadd73af33edb4fabc7da9d93c2f2d55c Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/149851288) Azure: [SUCCESS](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381)
* 090a72bd3101d7111490832526937d3973c75795 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] wsry commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
wsry commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r382404708
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -272,11 +287,161 @@ private static ShuffleDescriptor createShuffleDescriptor(
ResultPartitionID resultPartitionID,
ResourceID location,
TaskManagerLocation senderLocation,
- int channel) {
+ int connectionIndex) {
final NettyShuffleDescriptorBuilder builder = NettyShuffleDescriptorBuilder.newBuilder()
.setId(resultPartitionID)
.setProducerInfoFromTaskManagerLocation(senderLocation)
- .setConnectionIndex(channel);
+ .setConnectionIndex(connectionIndex);
return localMode ? builder.setProducerLocation(location).buildLocal() : builder.buildRemote();
}
+
+ /**
+ * A {@link SingleInputGateFactory} which replaces the default {@link RemoteInputChannel} and
+ * {@link LocalInputChannel} implementation with costume ones.
+ */
+ private static class TestSingleInputGateFactory extends SingleInputGateFactory {
+
+ private final ResourceID taskExecutorResourceId;
+ private final int partitionRequestInitialBackoff;
+ private final int partitionRequestMaxBackoff;
+ private final ConnectionManager connectionManager;
+ private final ResultPartitionManager partitionManager;
+ private final TaskEventPublisher taskEventPublisher;
+ private final NetworkBufferPool networkBufferPool;
+
+ public TestSingleInputGateFactory(
+ @Nonnull ResourceID taskExecutorResourceId,
+ @Nonnull NettyShuffleEnvironmentConfiguration networkConfig,
+ @Nonnull ConnectionManager connectionManager,
+ @Nonnull ResultPartitionManager partitionManager,
+ @Nonnull TaskEventPublisher taskEventPublisher,
+ @Nonnull NetworkBufferPool networkBufferPool) {
+ super(
+ taskExecutorResourceId,
+ networkConfig,
+ connectionManager,
+ partitionManager,
+ taskEventPublisher,
+ networkBufferPool);
+ this.networkBufferPool = networkBufferPool;
+ this.taskEventPublisher = taskEventPublisher;
+ this.partitionManager = partitionManager;
+ this.connectionManager = connectionManager;
+ this.partitionRequestMaxBackoff = networkConfig.partitionRequestMaxBackoff();
+ this.partitionRequestInitialBackoff = networkConfig.partitionRequestInitialBackoff();
+ this.taskExecutorResourceId = taskExecutorResourceId;
+ }
+
+ @Override
+ protected InputChannel createKnownInputChannel(
+ SingleInputGate inputGate,
+ int index,
+ NettyShuffleDescriptor inputChannelDescriptor,
+ ChannelStatistics channelStatistics,
+ InputChannelMetrics metrics) {
+ ResultPartitionID partitionId = inputChannelDescriptor.getResultPartitionID();
+ if (inputChannelDescriptor.isLocalTo(taskExecutorResourceId)) {
+ return new TestLocalInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics);
+ } else {
+ return new TestRemoteInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ inputChannelDescriptor.getConnectionId(),
+ connectionManager,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics,
+ networkBufferPool);
+ }
+ }
+ }
+
+ /**
+ * A {@link LocalInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestLocalInputChannel extends LocalInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
+
+ public TestLocalInputChannel(
+ SingleInputGate inputGate,
+ int channelIndex,
+ ResultPartitionID partitionId,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ int initialBackoff,
+ int maxBackoff,
+ InputChannelMetrics metrics) {
+ super(
+ inputGate,
+ channelIndex,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ initialBackoff,
+ maxBackoff,
+ metrics);
+ }
+
+ @Override
+ public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ super.requestSubpartition(channelIndex);
+ }
+
+ @Override
+ public ResultPartitionID getPartitionId() {
Review comment:
These channels consume the same result partition so with the same partition id. However, In SingleInputGate, IntermediateResultPartitionID is used as key of channel map, that is, SingleInputGate assumes that there is never two channels consume one result partition. See ```setInputChannel``` method in SingleInputGate.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-590753152
Also attach the comparison results executed in benchmark machine:
Baseline | |
-- | -- | --
DataSkewStreamNetworkThroughputBenchmarkExecutor| 51977.13475 |
StreamNetworkBroadcastThroughputBenchmarkExecutor| 1164.463375 |
StreamNetworkThroughputBenchmarkExecutor| 69553.22069 | 100,100ms
StreamNetworkThroughputBenchmarkExecutor| 17430.11747 | 100,100ms,SSL
StreamNetworkThroughputBenchmarkExecutor| 51019.51607 | 1000,1ms
StreamNetworkThroughputBenchmarkExecutor| 55423.23278 | 1000,100ms
StreamNetworkThroughputBenchmarkExecutor| 14924.65899 | 1000,100ms,SSL
less gates with less connections | |
DataSkewStreamNetworkThroughputBenchmarkExecutor| 33215.6767 |
StreamNetworkBroadcastThroughputBenchmarkExecutor| 1189.239694 |
StreamNetworkThroughputBenchmarkExecutor| 72450.69576 | 100,100ms
StreamNetworkThroughputBenchmarkExecutor| 17018.46785 | 100,100ms,SSL
StreamNetworkThroughputBenchmarkExecutor| 45829.91493 | 1000,1ms
StreamNetworkThroughputBenchmarkExecutor| 52943.20546 | 1000,100ms
StreamNetworkThroughputBenchmarkExecutor| 13762.96631 | 1000,100ms,SSL
less gates with same connections | |
DataSkewStreamNetworkThroughputBenchmarkExecutor| 32342.12887|
StreamNetworkBroadcastThroughputBenchmarkExecutor| 1136.44912 |
StreamNetworkThroughputBenchmarkExecutor| 75151.951622 | 100,100ms
StreamNetworkThroughputBenchmarkExecutor| 16878.544466 | 100,100ms,SSL
StreamNetworkThroughputBenchmarkExecutor| 47041.479598 | 1000,1ms
StreamNetworkThroughputBenchmarkExecutor| 52384.517795 | 1000,100ms
StreamNetworkThroughputBenchmarkExecutor| 13824.809088 | 1000,100ms,SSL
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r383094754
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/SingleInputGateBenchmarkFactory.java
##########
@@ -0,0 +1,173 @@
+/*
+ * 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.streaming.runtime.io.benchmark;
+
+import org.apache.flink.core.memory.MemorySegmentProvider;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.io.network.ConnectionID;
+import org.apache.flink.runtime.io.network.ConnectionManager;
+import org.apache.flink.runtime.io.network.TaskEventPublisher;
+import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+import org.apache.flink.runtime.io.network.metrics.InputChannelMetrics;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionManager;
+import org.apache.flink.runtime.io.network.partition.consumer.InputChannel;
+import org.apache.flink.runtime.io.network.partition.consumer.LocalInputChannel;
+import org.apache.flink.runtime.io.network.partition.consumer.RemoteInputChannel;
+import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate;
+import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGateFactory;
+import org.apache.flink.runtime.shuffle.NettyShuffleDescriptor;
+import org.apache.flink.runtime.taskmanager.NettyShuffleEnvironmentConfiguration;
+
+import java.io.IOException;
+
+/**
+ * A benchmark-specific input gate factory which overrides the respective methods of creating
+ * {@link RemoteInputChannel} and {@link LocalInputChannel} for requesting specific subpartitions.
+ */
+public class SingleInputGateBenchmarkFactory extends SingleInputGateFactory {
+
+ public SingleInputGateBenchmarkFactory(
+ ResourceID taskExecutorResourceId,
+ NettyShuffleEnvironmentConfiguration networkConfig,
+ ConnectionManager connectionManager,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ NetworkBufferPool networkBufferPool) {
+ super(
+ taskExecutorResourceId,
+ networkConfig,
+ connectionManager,
+ partitionManager,
+ taskEventPublisher,
+ networkBufferPool);
+ }
+
+ @Override
+ protected InputChannel createKnownInputChannel(
+ SingleInputGate inputGate,
+ int index,
+ NettyShuffleDescriptor inputChannelDescriptor,
+ SingleInputGateFactory.ChannelStatistics channelStatistics,
+ InputChannelMetrics metrics) {
+ ResultPartitionID partitionId = inputChannelDescriptor.getResultPartitionID();
+ if (inputChannelDescriptor.isLocalTo(taskExecutorResourceId)) {
+ return new TestLocalInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics);
+ } else {
+ return new TestRemoteInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ inputChannelDescriptor.getConnectionId(),
+ connectionManager,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics,
+ networkBufferPool);
+ }
+ }
+
+ /**
+ * A {@link LocalInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestLocalInputChannel extends LocalInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
+
+ public TestLocalInputChannel(
+ SingleInputGate inputGate,
+ int channelIndex,
+ ResultPartitionID partitionId,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ int initialBackoff,
+ int maxBackoff,
+ InputChannelMetrics metrics) {
+ super(
+ inputGate,
+ channelIndex,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ initialBackoff,
+ maxBackoff,
+ metrics);
+ }
+
+ @Override
+ public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ super.requestSubpartition(channelIndex);
+ }
+
+ @Override
+ public ResultPartitionID getPartitionId() {
+ return newPartitionID;
+ }
+ }
+
+ /**
+ * A {@link RemoteInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestRemoteInputChannel extends RemoteInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
Review comment:
nit: final
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:FAILURE URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359 TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:FAILURE URL:https://travis-ci.com/flink-ci/flink/builds/149795742 TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
Hash:2d7b20f97e41151ed741a3f6d93756b776b30dc9 Status:CANCELED URL:https://travis-ci.com/flink-ci/flink/builds/149835681 TriggerType:PUSH TriggerID:2d7b20f97e41151ed741a3f6d93756b776b30dc9
Hash:cfc691c781ae54d63ecaa7c282e6cb34f73425d8 Status:CANCELED URL:https://travis-ci.com/flink-ci/flink/builds/149842861 TriggerType:PUSH TriggerID:cfc691c781ae54d63ecaa7c282e6cb34f73425d8
Hash:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c Status:SUCCESS URL:https://travis-ci.com/flink-ci/flink/builds/149851288 TriggerType:PUSH TriggerID:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c
Hash:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c Status:SUCCESS URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381 TriggerType:PUSH TriggerID:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c
Hash:090a72bd3101d7111490832526937d3973c75795 Status:FAILURE URL:https://travis-ci.com/flink-ci/flink/builds/149946832 TriggerType:PUSH TriggerID:090a72bd3101d7111490832526937d3973c75795
Hash:8fe5e3d33d372b15e15cc52171ea8c3844c02ef9 Status:PENDING URL:https://travis-ci.com/flink-ci/flink/builds/149949398 TriggerType:PUSH TriggerID:8fe5e3d33d372b15e15cc52171ea8c3844c02ef9
Hash:8fe5e3d33d372b15e15cc52171ea8c3844c02ef9 Status:SUCCESS URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5400 TriggerType:PUSH TriggerID:8fe5e3d33d372b15e15cc52171ea8c3844c02ef9
-->
## CI report:
* fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149795742) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359)
* 2d7b20f97e41151ed741a3f6d93756b776b30dc9 Travis: [CANCELED](https://travis-ci.com/flink-ci/flink/builds/149835681)
* cfc691c781ae54d63ecaa7c282e6cb34f73425d8 Travis: [CANCELED](https://travis-ci.com/flink-ci/flink/builds/149842861)
* 21b36d4fadd73af33edb4fabc7da9d93c2f2d55c Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/149851288) Azure: [SUCCESS](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381)
* 090a72bd3101d7111490832526937d3973c75795 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149946832)
* 8fe5e3d33d372b15e15cc52171ea8c3844c02ef9 Travis: [PENDING](https://travis-ci.com/flink-ci/flink/builds/149949398) Azure: [SUCCESS](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5400)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW merged pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW merged pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r382394612
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -84,7 +98,7 @@
private int dataPort;
- private SingleInputGateFactory gateFactory;
+ private TestSingleInputGateFactory gateFactory;
Review comment:
no need change, we can still reference the parent `SingleInputGateFactory`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149795742",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149835681",
"triggerID" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"triggerType" : "PUSH"
}, {
"hash" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149842861",
"triggerID" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149851288",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "090a72bd3101d7111490832526937d3973c75795",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149946832",
"triggerID" : "090a72bd3101d7111490832526937d3973c75795",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149949398",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5400",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8fe5e3d33d372b15e15cc52171ea8c3844c02ef9 Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/149949398) Azure: [SUCCESS](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5400)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-590720137
Thanks for the updates @wsry .
As for the issue of connection index, I prefer to keeping it as previous value, then we can only measure the performance effect caused by gate/channel amount.
If we want to adjust the connection amount via `connectionIndex`, it should be a separate work to do in future if confirmed necessary.
If the performance regression still exists after getting ride of connection factor, caused by total buffers reduction as you analyzed above, I am a bit torn whether we should increase the default floating buffer setting for these benchmarks in order to keep the performance same as before. Otherwise we might have a bad/invalid curve to trace the performance changes in history term. WDYT? @pnowojski
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149795742",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149835681",
"triggerID" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"triggerType" : "PUSH"
}, {
"hash" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149842861",
"triggerID" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149851288",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "090a72bd3101d7111490832526937d3973c75795",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149946832",
"triggerID" : "090a72bd3101d7111490832526937d3973c75795",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149949398",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5400",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "07c2811b17d4ee4bb517b5b1a98d7a3bc91f2dea",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "07c2811b17d4ee4bb517b5b1a98d7a3bc91f2dea",
"triggerType" : "PUSH"
}, {
"hash" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/flink-ci/flink/builds/150266416",
"triggerID" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"triggerType" : "PUSH"
}, {
"hash" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5506",
"triggerID" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 07c2811b17d4ee4bb517b5b1a98d7a3bc91f2dea UNKNOWN
* 8f263fe38efaf89c0f001d3995d64e7bd46493b1 Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/150266416) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5506)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r383090336
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/SingleInputGateBenchmarkFactory.java
##########
@@ -0,0 +1,173 @@
+/*
+ * 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.streaming.runtime.io.benchmark;
+
+import org.apache.flink.core.memory.MemorySegmentProvider;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.io.network.ConnectionID;
+import org.apache.flink.runtime.io.network.ConnectionManager;
+import org.apache.flink.runtime.io.network.TaskEventPublisher;
+import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+import org.apache.flink.runtime.io.network.metrics.InputChannelMetrics;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionManager;
+import org.apache.flink.runtime.io.network.partition.consumer.InputChannel;
+import org.apache.flink.runtime.io.network.partition.consumer.LocalInputChannel;
+import org.apache.flink.runtime.io.network.partition.consumer.RemoteInputChannel;
+import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate;
+import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGateFactory;
+import org.apache.flink.runtime.shuffle.NettyShuffleDescriptor;
+import org.apache.flink.runtime.taskmanager.NettyShuffleEnvironmentConfiguration;
+
+import java.io.IOException;
+
+/**
+ * A benchmark-specific input gate factory which overrides the respective methods of creating
+ * {@link RemoteInputChannel} and {@link LocalInputChannel} for requesting specific subpartitions.
+ */
+public class SingleInputGateBenchmarkFactory extends SingleInputGateFactory {
+
+ public SingleInputGateBenchmarkFactory(
+ ResourceID taskExecutorResourceId,
+ NettyShuffleEnvironmentConfiguration networkConfig,
+ ConnectionManager connectionManager,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ NetworkBufferPool networkBufferPool) {
+ super(
+ taskExecutorResourceId,
+ networkConfig,
+ connectionManager,
+ partitionManager,
+ taskEventPublisher,
+ networkBufferPool);
+ }
+
+ @Override
+ protected InputChannel createKnownInputChannel(
+ SingleInputGate inputGate,
+ int index,
+ NettyShuffleDescriptor inputChannelDescriptor,
+ SingleInputGateFactory.ChannelStatistics channelStatistics,
+ InputChannelMetrics metrics) {
+ ResultPartitionID partitionId = inputChannelDescriptor.getResultPartitionID();
+ if (inputChannelDescriptor.isLocalTo(taskExecutorResourceId)) {
+ return new TestLocalInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics);
+ } else {
+ return new TestRemoteInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ inputChannelDescriptor.getConnectionId(),
+ connectionManager,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics,
+ networkBufferPool);
+ }
+ }
+
+ /**
+ * A {@link LocalInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestLocalInputChannel extends LocalInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
+
+ public TestLocalInputChannel(
+ SingleInputGate inputGate,
+ int channelIndex,
+ ResultPartitionID partitionId,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ int initialBackoff,
+ int maxBackoff,
+ InputChannelMetrics metrics) {
+ super(
+ inputGate,
+ channelIndex,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ initialBackoff,
+ maxBackoff,
+ metrics);
+ }
+
+ @Override
+ public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ super.requestSubpartition(channelIndex);
+ }
+
+ @Override
+ public ResultPartitionID getPartitionId() {
+ return newPartitionID;
+ }
+ }
+
+ /**
+ * A {@link RemoteInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestRemoteInputChannel extends RemoteInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
+
+ public TestRemoteInputChannel(
+ SingleInputGate inputGate,
+ int channelIndex,
+ ResultPartitionID partitionId,
+ ConnectionID connectionId,
+ ConnectionManager connectionManager,
+ int initialBackOff,
+ int maxBackoff,
+ InputChannelMetrics metrics,
+ MemorySegmentProvider memorySegmentProvider) {
+ super(
+ inputGate,
+ channelIndex,
+ partitionId,
+ connectionId,
+ connectionManager,
+ initialBackOff,
+ maxBackoff,
+ metrics,
+ memorySegmentProvider);
+ }
+
+ @Override
+ public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ super.requestSubpartition(channelIndex);
+ }
+
+ @Override
+ public ResultPartitionID getPartitionId() {
Review comment:
Ignore this because of [comment](https://github.com/apache/flink/pull/11155#discussion_r383090260)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r383088311
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/SingleInputGateBenchmarkFactory.java
##########
@@ -0,0 +1,173 @@
+/*
+ * 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.streaming.runtime.io.benchmark;
+
+import org.apache.flink.core.memory.MemorySegmentProvider;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.io.network.ConnectionID;
+import org.apache.flink.runtime.io.network.ConnectionManager;
+import org.apache.flink.runtime.io.network.TaskEventPublisher;
+import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+import org.apache.flink.runtime.io.network.metrics.InputChannelMetrics;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionManager;
+import org.apache.flink.runtime.io.network.partition.consumer.InputChannel;
+import org.apache.flink.runtime.io.network.partition.consumer.LocalInputChannel;
+import org.apache.flink.runtime.io.network.partition.consumer.RemoteInputChannel;
+import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate;
+import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGateFactory;
+import org.apache.flink.runtime.shuffle.NettyShuffleDescriptor;
+import org.apache.flink.runtime.taskmanager.NettyShuffleEnvironmentConfiguration;
+
+import java.io.IOException;
+
+/**
+ * A benchmark-specific input gate factory which overrides the respective methods of creating
+ * {@link RemoteInputChannel} and {@link LocalInputChannel} for requesting specific subpartitions.
+ */
+public class SingleInputGateBenchmarkFactory extends SingleInputGateFactory {
+
+ public SingleInputGateBenchmarkFactory(
+ ResourceID taskExecutorResourceId,
+ NettyShuffleEnvironmentConfiguration networkConfig,
+ ConnectionManager connectionManager,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ NetworkBufferPool networkBufferPool) {
+ super(
+ taskExecutorResourceId,
+ networkConfig,
+ connectionManager,
+ partitionManager,
+ taskEventPublisher,
+ networkBufferPool);
+ }
+
+ @Override
+ protected InputChannel createKnownInputChannel(
+ SingleInputGate inputGate,
+ int index,
+ NettyShuffleDescriptor inputChannelDescriptor,
+ SingleInputGateFactory.ChannelStatistics channelStatistics,
+ InputChannelMetrics metrics) {
+ ResultPartitionID partitionId = inputChannelDescriptor.getResultPartitionID();
+ if (inputChannelDescriptor.isLocalTo(taskExecutorResourceId)) {
+ return new TestLocalInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics);
+ } else {
+ return new TestRemoteInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ inputChannelDescriptor.getConnectionId(),
+ connectionManager,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics,
+ networkBufferPool);
+ }
+ }
+
+ /**
+ * A {@link LocalInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestLocalInputChannel extends LocalInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
+
+ public TestLocalInputChannel(
+ SingleInputGate inputGate,
+ int channelIndex,
+ ResultPartitionID partitionId,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ int initialBackoff,
+ int maxBackoff,
+ InputChannelMetrics metrics) {
+ super(
+ inputGate,
+ channelIndex,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ initialBackoff,
+ maxBackoff,
+ metrics);
+ }
+
+ @Override
+ public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ super.requestSubpartition(channelIndex);
+ }
+
+ @Override
+ public ResultPartitionID getPartitionId() {
+ return newPartitionID;
+ }
+ }
+
+ /**
+ * A {@link RemoteInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestRemoteInputChannel extends RemoteInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
+
+ public TestRemoteInputChannel(
+ SingleInputGate inputGate,
+ int channelIndex,
+ ResultPartitionID partitionId,
+ ConnectionID connectionId,
+ ConnectionManager connectionManager,
+ int initialBackOff,
+ int maxBackoff,
+ InputChannelMetrics metrics,
+ MemorySegmentProvider memorySegmentProvider) {
+ super(
+ inputGate,
+ channelIndex,
+ partitionId,
+ connectionId,
+ connectionManager,
+ initialBackOff,
+ maxBackoff,
+ metrics,
+ memorySegmentProvider);
+ }
+
+ @Override
+ public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ super.requestSubpartition(channelIndex);
+ }
+
+ @Override
+ public ResultPartitionID getPartitionId() {
Review comment:
nit: it is better to give some explanations why we need to override this method, otherwise it is not very easy understood by others.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r383090260
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -272,11 +287,161 @@ private static ShuffleDescriptor createShuffleDescriptor(
ResultPartitionID resultPartitionID,
ResourceID location,
TaskManagerLocation senderLocation,
- int channel) {
+ int connectionIndex) {
final NettyShuffleDescriptorBuilder builder = NettyShuffleDescriptorBuilder.newBuilder()
.setId(resultPartitionID)
.setProducerInfoFromTaskManagerLocation(senderLocation)
- .setConnectionIndex(channel);
+ .setConnectionIndex(connectionIndex);
return localMode ? builder.setProducerLocation(location).buildLocal() : builder.buildRemote();
}
+
+ /**
+ * A {@link SingleInputGateFactory} which replaces the default {@link RemoteInputChannel} and
+ * {@link LocalInputChannel} implementation with costume ones.
+ */
+ private static class TestSingleInputGateFactory extends SingleInputGateFactory {
+
+ private final ResourceID taskExecutorResourceId;
+ private final int partitionRequestInitialBackoff;
+ private final int partitionRequestMaxBackoff;
+ private final ConnectionManager connectionManager;
+ private final ResultPartitionManager partitionManager;
+ private final TaskEventPublisher taskEventPublisher;
+ private final NetworkBufferPool networkBufferPool;
+
+ public TestSingleInputGateFactory(
+ @Nonnull ResourceID taskExecutorResourceId,
+ @Nonnull NettyShuffleEnvironmentConfiguration networkConfig,
+ @Nonnull ConnectionManager connectionManager,
+ @Nonnull ResultPartitionManager partitionManager,
+ @Nonnull TaskEventPublisher taskEventPublisher,
+ @Nonnull NetworkBufferPool networkBufferPool) {
+ super(
+ taskExecutorResourceId,
+ networkConfig,
+ connectionManager,
+ partitionManager,
+ taskEventPublisher,
+ networkBufferPool);
+ this.networkBufferPool = networkBufferPool;
+ this.taskEventPublisher = taskEventPublisher;
+ this.partitionManager = partitionManager;
+ this.connectionManager = connectionManager;
+ this.partitionRequestMaxBackoff = networkConfig.partitionRequestMaxBackoff();
+ this.partitionRequestInitialBackoff = networkConfig.partitionRequestInitialBackoff();
+ this.taskExecutorResourceId = taskExecutorResourceId;
+ }
+
+ @Override
+ protected InputChannel createKnownInputChannel(
+ SingleInputGate inputGate,
+ int index,
+ NettyShuffleDescriptor inputChannelDescriptor,
+ ChannelStatistics channelStatistics,
+ InputChannelMetrics metrics) {
+ ResultPartitionID partitionId = inputChannelDescriptor.getResultPartitionID();
+ if (inputChannelDescriptor.isLocalTo(taskExecutorResourceId)) {
+ return new TestLocalInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics);
+ } else {
+ return new TestRemoteInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ inputChannelDescriptor.getConnectionId(),
+ connectionManager,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics,
+ networkBufferPool);
+ }
+ }
+ }
+
+ /**
+ * A {@link LocalInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestLocalInputChannel extends LocalInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
+
+ public TestLocalInputChannel(
+ SingleInputGate inputGate,
+ int channelIndex,
+ ResultPartitionID partitionId,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ int initialBackoff,
+ int maxBackoff,
+ InputChannelMetrics metrics) {
+ super(
+ inputGate,
+ channelIndex,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ initialBackoff,
+ maxBackoff,
+ metrics);
+ }
+
+ @Override
+ public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ super.requestSubpartition(channelIndex);
+ }
+
+ @Override
+ public ResultPartitionID getPartitionId() {
Review comment:
I have some basic thoughts for the concerns of `ResultPartitionID` AND `subpartitionIndex`.
The baseline is that we can not break the existing rules that `ResultPartitionID` is different for every input channel but `subpartitionIndex` is same for all.
- `ResultPartitionID`: We can pass a desired value directly while constructing the `InputChannel` instance instead of getting it from `NettyShuffleDescriptor` now. Although it is still not consistent between `NettyShuffleDescriptor` and `InputChannel`, it becomes consistent in all the places inside channel, and we also avoid hacking the override method of `getPartitionId` in `TestRemoteInputChannel` below.
- `subpartitionIndex`: We can keep it same as now. Although we can also maintain this info in the `InputChannel` layer as did above, then we can also pass the desired subpartition index while constructing the `InputChannel` to further void hack `requestSubpartition` method. But considering it is same for all the channels in normal case, so it might be more reasonable to maintain it in gate layer as now.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r382390260
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -272,11 +287,161 @@ private static ShuffleDescriptor createShuffleDescriptor(
ResultPartitionID resultPartitionID,
ResourceID location,
TaskManagerLocation senderLocation,
- int channel) {
+ int connectionIndex) {
final NettyShuffleDescriptorBuilder builder = NettyShuffleDescriptorBuilder.newBuilder()
.setId(resultPartitionID)
.setProducerInfoFromTaskManagerLocation(senderLocation)
- .setConnectionIndex(channel);
+ .setConnectionIndex(connectionIndex);
return localMode ? builder.setProducerLocation(location).buildLocal() : builder.buildRemote();
}
+
+ /**
+ * A {@link SingleInputGateFactory} which replaces the default {@link RemoteInputChannel} and
+ * {@link LocalInputChannel} implementation with costume ones.
+ */
+ private static class TestSingleInputGateFactory extends SingleInputGateFactory {
+
+ private final ResourceID taskExecutorResourceId;
+ private final int partitionRequestInitialBackoff;
+ private final int partitionRequestMaxBackoff;
+ private final ConnectionManager connectionManager;
+ private final ResultPartitionManager partitionManager;
+ private final TaskEventPublisher taskEventPublisher;
+ private final NetworkBufferPool networkBufferPool;
+
+ public TestSingleInputGateFactory(
+ @Nonnull ResourceID taskExecutorResourceId,
+ @Nonnull NettyShuffleEnvironmentConfiguration networkConfig,
+ @Nonnull ConnectionManager connectionManager,
+ @Nonnull ResultPartitionManager partitionManager,
+ @Nonnull TaskEventPublisher taskEventPublisher,
+ @Nonnull NetworkBufferPool networkBufferPool) {
+ super(
+ taskExecutorResourceId,
+ networkConfig,
+ connectionManager,
+ partitionManager,
+ taskEventPublisher,
+ networkBufferPool);
+ this.networkBufferPool = networkBufferPool;
+ this.taskEventPublisher = taskEventPublisher;
+ this.partitionManager = partitionManager;
+ this.connectionManager = connectionManager;
+ this.partitionRequestMaxBackoff = networkConfig.partitionRequestMaxBackoff();
+ this.partitionRequestInitialBackoff = networkConfig.partitionRequestInitialBackoff();
+ this.taskExecutorResourceId = taskExecutorResourceId;
+ }
+
+ @Override
+ protected InputChannel createKnownInputChannel(
+ SingleInputGate inputGate,
+ int index,
+ NettyShuffleDescriptor inputChannelDescriptor,
+ ChannelStatistics channelStatistics,
+ InputChannelMetrics metrics) {
+ ResultPartitionID partitionId = inputChannelDescriptor.getResultPartitionID();
+ if (inputChannelDescriptor.isLocalTo(taskExecutorResourceId)) {
+ return new TestLocalInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics);
+ } else {
+ return new TestRemoteInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ inputChannelDescriptor.getConnectionId(),
+ connectionManager,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics,
+ networkBufferPool);
+ }
+ }
+ }
+
+ /**
+ * A {@link LocalInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestLocalInputChannel extends LocalInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
+
+ public TestLocalInputChannel(
+ SingleInputGate inputGate,
+ int channelIndex,
+ ResultPartitionID partitionId,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ int initialBackoff,
+ int maxBackoff,
+ InputChannelMetrics metrics) {
+ super(
+ inputGate,
+ channelIndex,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ initialBackoff,
+ maxBackoff,
+ metrics);
+ }
+
+ @Override
+ public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ super.requestSubpartition(channelIndex);
+ }
+
+ @Override
+ public ResultPartitionID getPartitionId() {
Review comment:
Why we need to override this method? I guess that the original `partitionId` while constructing the `SingleInputGate` is valid.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149795742",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149835681",
"triggerID" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"triggerType" : "PUSH"
}, {
"hash" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149842861",
"triggerID" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149851288",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "090a72bd3101d7111490832526937d3973c75795",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149946832",
"triggerID" : "090a72bd3101d7111490832526937d3973c75795",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149949398",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5400",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/150266416",
"triggerID" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"triggerType" : "PUSH"
}, {
"hash" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5506",
"triggerID" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"triggerType" : "PUSH"
}, {
"hash" : "fbf92243b17619df704bf2600d294cda449c7ddc",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/flink-ci/flink/builds/150577037",
"triggerID" : "fbf92243b17619df704bf2600d294cda449c7ddc",
"triggerType" : "PUSH"
}, {
"hash" : "035041ed96b561f0d7bb69fa09436a24e42a5a54",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "035041ed96b561f0d7bb69fa09436a24e42a5a54",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* fbf92243b17619df704bf2600d294cda449c7ddc Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/150577037)
* 035041ed96b561f0d7bb69fa09436a24e42a5a54 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r383090260
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -272,11 +287,161 @@ private static ShuffleDescriptor createShuffleDescriptor(
ResultPartitionID resultPartitionID,
ResourceID location,
TaskManagerLocation senderLocation,
- int channel) {
+ int connectionIndex) {
final NettyShuffleDescriptorBuilder builder = NettyShuffleDescriptorBuilder.newBuilder()
.setId(resultPartitionID)
.setProducerInfoFromTaskManagerLocation(senderLocation)
- .setConnectionIndex(channel);
+ .setConnectionIndex(connectionIndex);
return localMode ? builder.setProducerLocation(location).buildLocal() : builder.buildRemote();
}
+
+ /**
+ * A {@link SingleInputGateFactory} which replaces the default {@link RemoteInputChannel} and
+ * {@link LocalInputChannel} implementation with costume ones.
+ */
+ private static class TestSingleInputGateFactory extends SingleInputGateFactory {
+
+ private final ResourceID taskExecutorResourceId;
+ private final int partitionRequestInitialBackoff;
+ private final int partitionRequestMaxBackoff;
+ private final ConnectionManager connectionManager;
+ private final ResultPartitionManager partitionManager;
+ private final TaskEventPublisher taskEventPublisher;
+ private final NetworkBufferPool networkBufferPool;
+
+ public TestSingleInputGateFactory(
+ @Nonnull ResourceID taskExecutorResourceId,
+ @Nonnull NettyShuffleEnvironmentConfiguration networkConfig,
+ @Nonnull ConnectionManager connectionManager,
+ @Nonnull ResultPartitionManager partitionManager,
+ @Nonnull TaskEventPublisher taskEventPublisher,
+ @Nonnull NetworkBufferPool networkBufferPool) {
+ super(
+ taskExecutorResourceId,
+ networkConfig,
+ connectionManager,
+ partitionManager,
+ taskEventPublisher,
+ networkBufferPool);
+ this.networkBufferPool = networkBufferPool;
+ this.taskEventPublisher = taskEventPublisher;
+ this.partitionManager = partitionManager;
+ this.connectionManager = connectionManager;
+ this.partitionRequestMaxBackoff = networkConfig.partitionRequestMaxBackoff();
+ this.partitionRequestInitialBackoff = networkConfig.partitionRequestInitialBackoff();
+ this.taskExecutorResourceId = taskExecutorResourceId;
+ }
+
+ @Override
+ protected InputChannel createKnownInputChannel(
+ SingleInputGate inputGate,
+ int index,
+ NettyShuffleDescriptor inputChannelDescriptor,
+ ChannelStatistics channelStatistics,
+ InputChannelMetrics metrics) {
+ ResultPartitionID partitionId = inputChannelDescriptor.getResultPartitionID();
+ if (inputChannelDescriptor.isLocalTo(taskExecutorResourceId)) {
+ return new TestLocalInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics);
+ } else {
+ return new TestRemoteInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ inputChannelDescriptor.getConnectionId(),
+ connectionManager,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics,
+ networkBufferPool);
+ }
+ }
+ }
+
+ /**
+ * A {@link LocalInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestLocalInputChannel extends LocalInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
+
+ public TestLocalInputChannel(
+ SingleInputGate inputGate,
+ int channelIndex,
+ ResultPartitionID partitionId,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ int initialBackoff,
+ int maxBackoff,
+ InputChannelMetrics metrics) {
+ super(
+ inputGate,
+ channelIndex,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ initialBackoff,
+ maxBackoff,
+ metrics);
+ }
+
+ @Override
+ public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ super.requestSubpartition(channelIndex);
+ }
+
+ @Override
+ public ResultPartitionID getPartitionId() {
Review comment:
I have some basic thoughts for the concerns of `ResultPartitionID` and `subpartitionIndex`.
The baseline is that we can not break the existing rules that `ResultPartitionID` is different for every input channel but `subpartitionIndex` is same for all.
- `ResultPartitionID`: We can pass a desired value directly while constructing the `InputChannel` instance instead of getting it from `NettyShuffleDescriptor` now. Although it is still not consistent between `NettyShuffleDescriptor` and `InputChannel`, it becomes consistent in all the places inside channel, and we also avoid hacking the override method of `getPartitionId` in `TestRemoteInputChannel` below.
- `subpartitionIndex`: We can keep it same as now. Although we can also maintain this info in the `InputChannel` layer as did above, then we can also pass the desired subpartition index while constructing the `InputChannel` to further void hack `requestSubpartition` method. But considering it is same for all the channels in normal case, so it might be more reasonable to maintain it in gate layer as now.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r382441933
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
##########
@@ -95,7 +95,7 @@ public LocalInputChannel(
// ------------------------------------------------------------------------
Review comment:
Please copy PR description into the commit message and once you have some expected benchmark results, please include them in the commit message as well :)
Could you also explain in both places, how are you solving the problem? Maybe something along those lines:
```
Before this change, in network benchmark (for example 1000 channels benchmark with 4 record writers) StreamNetworkBenchmarkEnvironment#createInputGate was creating 1000 input gates with 4 input channels each, which doesn't make much sense. This commit is changing that to a single receiver with 4 input gates, with 1000 channels each.
It is achieved by providing testing implementations of InputChannels, which are using channel index for requesting subpartitions from ResultPartition, instead of subpartition index. Thanks to that, we can connect single ResultPartition with N subpartitions, to a single instance of InputGate with 1000 channels.
```
?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:FAILURE URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359 TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:FAILURE URL:https://travis-ci.com/flink-ci/flink/builds/149795742 TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
Hash:2d7b20f97e41151ed741a3f6d93756b776b30dc9 Status:PENDING URL:https://travis-ci.com/flink-ci/flink/builds/149835681 TriggerType:PUSH TriggerID:2d7b20f97e41151ed741a3f6d93756b776b30dc9
Hash:cfc691c781ae54d63ecaa7c282e6cb34f73425d8 Status:UNKNOWN URL:TBD TriggerType:PUSH TriggerID:cfc691c781ae54d63ecaa7c282e6cb34f73425d8
-->
## CI report:
* fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149795742) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359)
* 2d7b20f97e41151ed741a3f6d93756b776b30dc9 Travis: [PENDING](https://travis-ci.com/flink-ci/flink/builds/149835681)
* cfc691c781ae54d63ecaa7c282e6cb34f73425d8 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149795742",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149835681",
"triggerID" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"triggerType" : "PUSH"
}, {
"hash" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149842861",
"triggerID" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149851288",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "090a72bd3101d7111490832526937d3973c75795",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149946832",
"triggerID" : "090a72bd3101d7111490832526937d3973c75795",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149949398",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5400",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/flink-ci/flink/builds/150266416",
"triggerID" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"triggerType" : "PUSH"
}, {
"hash" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5506",
"triggerID" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"triggerType" : "PUSH"
}, {
"hash" : "fbf92243b17619df704bf2600d294cda449c7ddc",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "fbf92243b17619df704bf2600d294cda449c7ddc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8f263fe38efaf89c0f001d3995d64e7bd46493b1 Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/150266416) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5506)
* fbf92243b17619df704bf2600d294cda449c7ddc UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "FAILURE",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149795742",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"status" : "CANCELED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149835681",
"triggerID" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"triggerType" : "PUSH"
}, {
"hash" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"status" : "CANCELED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149842861",
"triggerID" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149851288",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "090a72bd3101d7111490832526937d3973c75795",
"status" : "FAILURE",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149946832",
"triggerID" : "090a72bd3101d7111490832526937d3973c75795",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149949398",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5400",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149795742) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359)
* 2d7b20f97e41151ed741a3f6d93756b776b30dc9 Travis: [CANCELED](https://travis-ci.com/flink-ci/flink/builds/149835681)
* cfc691c781ae54d63ecaa7c282e6cb34f73425d8 Travis: [CANCELED](https://travis-ci.com/flink-ci/flink/builds/149842861)
* 21b36d4fadd73af33edb4fabc7da9d93c2f2d55c Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/149851288) Azure: [SUCCESS](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381)
* 090a72bd3101d7111490832526937d3973c75795 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149946832)
* 8fe5e3d33d372b15e15cc52171ea8c3844c02ef9 Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/149949398) Azure: [SUCCESS](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5400)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] wsry commented on issue #11155: [FLINK-14818][benchmark]
Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
wsry commented on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-590217451
I ran the micro benchmark and found that some of the performance results become better and some go worse. The results are as follows:
Before the change:
```
Benchmark (channelsFlushTimeout) Mode Cnt Score Error Units
DataSkewStreamNetworkThroughputBenchmarkExecutor.networkSkewedThroughput N/A thrpt 30 17079.534 ± 830.532 ops/ms
StreamNetworkBroadcastThroughputBenchmarkExecutor.networkBroadcastThroughput N/A thrpt 30 599.664 ± 13.325 ops/ms
StreamNetworkThroughputBenchmarkExecutor.networkThroughput 100,100ms thrpt 30 45629.898 ± 1623.455 ops/ms
StreamNetworkThroughputBenchmarkExecutor.networkThroughput 100,100ms,SSL thrpt 30 9817.421 ± 216.075 ops/ms
StreamNetworkThroughputBenchmarkExecutor.networkThroughput 1000,1ms thrpt 30 25442.152 ± 968.340 ops/ms
StreamNetworkThroughputBenchmarkExecutor.networkThroughput 1000,100ms thrpt 30 27944.285 ± 518.106 ops/ms
StreamNetworkThroughputBenchmarkExecutor.networkThroughput 1000,100ms,SSL thrpt 30 7820.549 ± 895.862 ops/ms
StreamNetworkLatencyBenchmarkExecutor.networkLatency1to1 N/A avgt 30 13.184 ± 0.093 ms/op
```
And after the change:
```
Benchmark (channelsFlushTimeout) Mode Cnt Score Error Units
DataSkewStreamNetworkThroughputBenchmarkExecutor.networkSkewedThroughput N/A thrpt 30 22758.888 ± 613.743 ops/ms
StreamNetworkBroadcastThroughputBenchmarkExecutor.networkBroadcastThroughput N/A thrpt 30 634.810 ± 14.375 ops/ms
StreamNetworkThroughputBenchmarkExecutor.networkThroughput 100,100ms thrpt 30 32589.803 ± 3150.003 ops/ms
StreamNetworkThroughputBenchmarkExecutor.networkThroughput 100,100ms,SSL thrpt 30 9966.284 ± 296.523 ops/ms
StreamNetworkThroughputBenchmarkExecutor.networkThroughput 1000,1ms thrpt 30 26633.166 ± 1123.461 ops/ms
StreamNetworkThroughputBenchmarkExecutor.networkThroughput 1000,100ms thrpt 30 29845.823 ± 564.253 ops/ms
StreamNetworkThroughputBenchmarkExecutor.networkThroughput 1000,100ms,SSL thrpt 30 8740.787 ± 184.125 ops/ms
StreamNetworkLatencyBenchmarkExecutor.networkLatency1to1 N/A avgt 30 13.047 ± 0.155 ms/op
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot commented on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot commented on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588915466
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit fee93cecf80cb73fcdf082f47fe0b9bd727a3002 (Thu Feb 20 10:25:48 UTC 2020)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r383090816
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -272,11 +287,161 @@ private static ShuffleDescriptor createShuffleDescriptor(
ResultPartitionID resultPartitionID,
ResourceID location,
TaskManagerLocation senderLocation,
- int channel) {
+ int connectionIndex) {
final NettyShuffleDescriptorBuilder builder = NettyShuffleDescriptorBuilder.newBuilder()
.setId(resultPartitionID)
.setProducerInfoFromTaskManagerLocation(senderLocation)
- .setConnectionIndex(channel);
+ .setConnectionIndex(connectionIndex);
return localMode ? builder.setProducerLocation(location).buildLocal() : builder.buildRemote();
}
+
+ /**
+ * A {@link SingleInputGateFactory} which replaces the default {@link RemoteInputChannel} and
+ * {@link LocalInputChannel} implementation with costume ones.
+ */
+ private static class TestSingleInputGateFactory extends SingleInputGateFactory {
+
+ private final ResourceID taskExecutorResourceId;
+ private final int partitionRequestInitialBackoff;
+ private final int partitionRequestMaxBackoff;
+ private final ConnectionManager connectionManager;
+ private final ResultPartitionManager partitionManager;
+ private final TaskEventPublisher taskEventPublisher;
+ private final NetworkBufferPool networkBufferPool;
+
+ public TestSingleInputGateFactory(
+ @Nonnull ResourceID taskExecutorResourceId,
+ @Nonnull NettyShuffleEnvironmentConfiguration networkConfig,
+ @Nonnull ConnectionManager connectionManager,
+ @Nonnull ResultPartitionManager partitionManager,
+ @Nonnull TaskEventPublisher taskEventPublisher,
+ @Nonnull NetworkBufferPool networkBufferPool) {
+ super(
+ taskExecutorResourceId,
+ networkConfig,
+ connectionManager,
+ partitionManager,
+ taskEventPublisher,
+ networkBufferPool);
+ this.networkBufferPool = networkBufferPool;
+ this.taskEventPublisher = taskEventPublisher;
+ this.partitionManager = partitionManager;
+ this.connectionManager = connectionManager;
+ this.partitionRequestMaxBackoff = networkConfig.partitionRequestMaxBackoff();
+ this.partitionRequestInitialBackoff = networkConfig.partitionRequestInitialBackoff();
+ this.taskExecutorResourceId = taskExecutorResourceId;
+ }
+
+ @Override
+ protected InputChannel createKnownInputChannel(
+ SingleInputGate inputGate,
+ int index,
+ NettyShuffleDescriptor inputChannelDescriptor,
+ ChannelStatistics channelStatistics,
+ InputChannelMetrics metrics) {
+ ResultPartitionID partitionId = inputChannelDescriptor.getResultPartitionID();
+ if (inputChannelDescriptor.isLocalTo(taskExecutorResourceId)) {
+ return new TestLocalInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics);
+ } else {
+ return new TestRemoteInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ inputChannelDescriptor.getConnectionId(),
+ connectionManager,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics,
+ networkBufferPool);
+ }
+ }
+ }
+
+ /**
+ * A {@link LocalInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestLocalInputChannel extends LocalInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
+
+ public TestLocalInputChannel(
+ SingleInputGate inputGate,
+ int channelIndex,
+ ResultPartitionID partitionId,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ int initialBackoff,
+ int maxBackoff,
+ InputChannelMetrics metrics) {
+ super(
+ inputGate,
+ channelIndex,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ initialBackoff,
+ maxBackoff,
+ metrics);
+ }
+
+ @Override
+ public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ super.requestSubpartition(channelIndex);
+ }
+
+ @Override
+ public ResultPartitionID getPartitionId() {
Review comment:
In addition, while reviewing the current usage of `InputChannel#getPartitionId`, I find it is mainly used for generating `AddCredit` message. But the `ResultPartitionID` was actually not used for upstream side in practice.
So I prefer to remove this field from `AddCredit` message, which can further reduce the message size from current 52 bytes to 20 bytes.
@pnowojski do you have other concerns for removing it? If not, i would create a separate ticket for this issue.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-589501898
@pnowojski I guess you might also be interested in taking a look. :)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] wsry commented on issue #11155: [FLINK-14818][benchmark]
Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
wsry commented on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-590666593
I find that the number of tcp connections is one of the factors which influence the performance. If I change the connectionIndex from gateIndex to channelIndex which means it remains unchanged compared with the behavior before this change (setup 1000 tcp connections per gate if the number of channels is 1000), the performance results are as follows. From the results, we can see that the performance goes down a bit compared with the results before this change, which should be caused by the decrease of floating buffers. Then what do you think? Should we keep the number of connections unchanged in this PR? @zhijiangW @pnowojski
```
Benchmark (channelsFlushTimeout) Mode Cnt Score Error Units
DataSkewStreamNetworkThroughputBenchmarkExecutor.networkSkewedThroughput N/A thrpt 30 17345.574 ± 370.647 ops/ms
StreamNetworkBroadcastThroughputBenchmarkExecutor.networkBroadcastThroughput N/A thrpt 30 608.881 ± 12.054 ops/ms
StreamNetworkThroughputBenchmarkExecutor.networkThroughput 100,100ms thrpt 30 41732.518 ± 1109.436 ops/ms
StreamNetworkThroughputBenchmarkExecutor.networkThroughput 100,100ms,SSL thrpt 30 9689.525 ± 202.895 ops/ms
StreamNetworkThroughputBenchmarkExecutor.networkThroughput 1000,1ms thrpt 30 24106.705 ± 2952.364 ops/ms
StreamNetworkThroughputBenchmarkExecutor.networkThroughput 1000,100ms thrpt 30 27509.665 ± 3246.965 ops/ms
StreamNetworkThroughputBenchmarkExecutor.networkThroughput 1000,100ms,SSL thrpt 30 7691.287 ± 927.775 ops/ms
StreamNetworkLatencyBenchmarkExecutor.networkLatency1to1 N/A avgt 30 12.758 ± 0.147 ms/op
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r382395379
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateFactory.java
##########
@@ -243,7 +244,10 @@ private InputChannel createKnownInputChannel(
return () -> bufferPoolFactory.createBufferPool(0, floatingNetworkBuffersPerGate);
}
- private static class ChannelStatistics {
+ /**
+ * Statistics of input channels.
Review comment:
nit: Not really necessary change, because it does not provide any rich informations than the class name.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] wsry commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
wsry commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r382418406
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateFactory.java
##########
@@ -243,7 +244,10 @@ private InputChannel createKnownInputChannel(
return () -> bufferPoolFactory.createBufferPool(0, floatingNetworkBuffersPerGate);
}
- private static class ChannelStatistics {
+ /**
+ * Statistics of input channels.
Review comment:
The comment is only for style checking. Changing the class from package private to protected fails the style checking.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149795742",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149835681",
"triggerID" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"triggerType" : "PUSH"
}, {
"hash" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149842861",
"triggerID" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149851288",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "090a72bd3101d7111490832526937d3973c75795",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149946832",
"triggerID" : "090a72bd3101d7111490832526937d3973c75795",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149949398",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5400",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/150266416",
"triggerID" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"triggerType" : "PUSH"
}, {
"hash" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5506",
"triggerID" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"triggerType" : "PUSH"
}, {
"hash" : "fbf92243b17619df704bf2600d294cda449c7ddc",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/150577037",
"triggerID" : "fbf92243b17619df704bf2600d294cda449c7ddc",
"triggerType" : "PUSH"
}, {
"hash" : "035041ed96b561f0d7bb69fa09436a24e42a5a54",
"status" : "FAILURE",
"url" : "https://travis-ci.com/flink-ci/flink/builds/150774056",
"triggerID" : "035041ed96b561f0d7bb69fa09436a24e42a5a54",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 035041ed96b561f0d7bb69fa09436a24e42a5a54 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/150774056)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r383102071
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -272,11 +287,161 @@ private static ShuffleDescriptor createShuffleDescriptor(
ResultPartitionID resultPartitionID,
ResourceID location,
TaskManagerLocation senderLocation,
- int channel) {
+ int connectionIndex) {
final NettyShuffleDescriptorBuilder builder = NettyShuffleDescriptorBuilder.newBuilder()
.setId(resultPartitionID)
.setProducerInfoFromTaskManagerLocation(senderLocation)
- .setConnectionIndex(channel);
+ .setConnectionIndex(connectionIndex);
return localMode ? builder.setProducerLocation(location).buildLocal() : builder.buildRemote();
}
+
+ /**
+ * A {@link SingleInputGateFactory} which replaces the default {@link RemoteInputChannel} and
+ * {@link LocalInputChannel} implementation with costume ones.
+ */
+ private static class TestSingleInputGateFactory extends SingleInputGateFactory {
+
+ private final ResourceID taskExecutorResourceId;
+ private final int partitionRequestInitialBackoff;
+ private final int partitionRequestMaxBackoff;
+ private final ConnectionManager connectionManager;
+ private final ResultPartitionManager partitionManager;
+ private final TaskEventPublisher taskEventPublisher;
+ private final NetworkBufferPool networkBufferPool;
+
+ public TestSingleInputGateFactory(
+ @Nonnull ResourceID taskExecutorResourceId,
+ @Nonnull NettyShuffleEnvironmentConfiguration networkConfig,
+ @Nonnull ConnectionManager connectionManager,
+ @Nonnull ResultPartitionManager partitionManager,
+ @Nonnull TaskEventPublisher taskEventPublisher,
+ @Nonnull NetworkBufferPool networkBufferPool) {
+ super(
+ taskExecutorResourceId,
+ networkConfig,
+ connectionManager,
+ partitionManager,
+ taskEventPublisher,
+ networkBufferPool);
+ this.networkBufferPool = networkBufferPool;
+ this.taskEventPublisher = taskEventPublisher;
+ this.partitionManager = partitionManager;
+ this.connectionManager = connectionManager;
+ this.partitionRequestMaxBackoff = networkConfig.partitionRequestMaxBackoff();
+ this.partitionRequestInitialBackoff = networkConfig.partitionRequestInitialBackoff();
+ this.taskExecutorResourceId = taskExecutorResourceId;
+ }
+
+ @Override
+ protected InputChannel createKnownInputChannel(
+ SingleInputGate inputGate,
+ int index,
+ NettyShuffleDescriptor inputChannelDescriptor,
+ ChannelStatistics channelStatistics,
+ InputChannelMetrics metrics) {
+ ResultPartitionID partitionId = inputChannelDescriptor.getResultPartitionID();
+ if (inputChannelDescriptor.isLocalTo(taskExecutorResourceId)) {
+ return new TestLocalInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics);
+ } else {
+ return new TestRemoteInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ inputChannelDescriptor.getConnectionId(),
+ connectionManager,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics,
+ networkBufferPool);
+ }
+ }
+ }
+
+ /**
+ * A {@link LocalInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestLocalInputChannel extends LocalInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
+
+ public TestLocalInputChannel(
+ SingleInputGate inputGate,
+ int channelIndex,
+ ResultPartitionID partitionId,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ int initialBackoff,
+ int maxBackoff,
+ InputChannelMetrics metrics) {
+ super(
+ inputGate,
+ channelIndex,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ initialBackoff,
+ maxBackoff,
+ metrics);
+ }
+
+ @Override
+ public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ super.requestSubpartition(channelIndex);
+ }
+
+ @Override
+ public ResultPartitionID getPartitionId() {
Review comment:
I forgot the case of requesting partition via `ResultPartitionID`, so the above option for `ResultPartitionID` is invalid.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:FAILURE URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359 TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:FAILURE URL:https://travis-ci.com/flink-ci/flink/builds/149795742 TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
Hash:2d7b20f97e41151ed741a3f6d93756b776b30dc9 Status:CANCELED URL:https://travis-ci.com/flink-ci/flink/builds/149835681 TriggerType:PUSH TriggerID:2d7b20f97e41151ed741a3f6d93756b776b30dc9
Hash:cfc691c781ae54d63ecaa7c282e6cb34f73425d8 Status:CANCELED URL:https://travis-ci.com/flink-ci/flink/builds/149842861 TriggerType:PUSH TriggerID:cfc691c781ae54d63ecaa7c282e6cb34f73425d8
Hash:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c Status:SUCCESS URL:https://travis-ci.com/flink-ci/flink/builds/149851288 TriggerType:PUSH TriggerID:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c
Hash:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c Status:SUCCESS URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381 TriggerType:PUSH TriggerID:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c
-->
## CI report:
* fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149795742) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359)
* 2d7b20f97e41151ed741a3f6d93756b776b30dc9 Travis: [CANCELED](https://travis-ci.com/flink-ci/flink/builds/149835681)
* cfc691c781ae54d63ecaa7c282e6cb34f73425d8 Travis: [CANCELED](https://travis-ci.com/flink-ci/flink/builds/149842861)
* 21b36d4fadd73af33edb4fabc7da9d93c2f2d55c Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/149851288) Azure: [SUCCESS](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:FAILURE URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359 TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:FAILURE URL:https://travis-ci.com/flink-ci/flink/builds/149795742 TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
Hash:2d7b20f97e41151ed741a3f6d93756b776b30dc9 Status:CANCELED URL:https://travis-ci.com/flink-ci/flink/builds/149835681 TriggerType:PUSH TriggerID:2d7b20f97e41151ed741a3f6d93756b776b30dc9
Hash:cfc691c781ae54d63ecaa7c282e6cb34f73425d8 Status:CANCELED URL:https://travis-ci.com/flink-ci/flink/builds/149842861 TriggerType:PUSH TriggerID:cfc691c781ae54d63ecaa7c282e6cb34f73425d8
Hash:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c Status:SUCCESS URL:https://travis-ci.com/flink-ci/flink/builds/149851288 TriggerType:PUSH TriggerID:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c
Hash:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c Status:SUCCESS URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381 TriggerType:PUSH TriggerID:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c
Hash:090a72bd3101d7111490832526937d3973c75795 Status:FAILURE URL:https://travis-ci.com/flink-ci/flink/builds/149946832 TriggerType:PUSH TriggerID:090a72bd3101d7111490832526937d3973c75795
Hash:8fe5e3d33d372b15e15cc52171ea8c3844c02ef9 Status:UNKNOWN URL:TBD TriggerType:PUSH TriggerID:8fe5e3d33d372b15e15cc52171ea8c3844c02ef9
-->
## CI report:
* fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149795742) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359)
* 2d7b20f97e41151ed741a3f6d93756b776b30dc9 Travis: [CANCELED](https://travis-ci.com/flink-ci/flink/builds/149835681)
* cfc691c781ae54d63ecaa7c282e6cb34f73425d8 Travis: [CANCELED](https://travis-ci.com/flink-ci/flink/builds/149842861)
* 21b36d4fadd73af33edb4fabc7da9d93c2f2d55c Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/149851288) Azure: [SUCCESS](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381)
* 090a72bd3101d7111490832526937d3973c75795 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149946832)
* 8fe5e3d33d372b15e15cc52171ea8c3844c02ef9 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r382395379
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateFactory.java
##########
@@ -243,7 +244,10 @@ private InputChannel createKnownInputChannel(
return () -> bufferPoolFactory.createBufferPool(0, floatingNetworkBuffersPerGate);
}
- private static class ChannelStatistics {
+ /**
+ * Statistics of input channels.
Review comment:
nit: Not really necessary because it does not provide any rich informations than the class name.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r382389631
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -272,11 +287,161 @@ private static ShuffleDescriptor createShuffleDescriptor(
ResultPartitionID resultPartitionID,
ResourceID location,
TaskManagerLocation senderLocation,
- int channel) {
+ int connectionIndex) {
final NettyShuffleDescriptorBuilder builder = NettyShuffleDescriptorBuilder.newBuilder()
.setId(resultPartitionID)
.setProducerInfoFromTaskManagerLocation(senderLocation)
- .setConnectionIndex(channel);
+ .setConnectionIndex(connectionIndex);
return localMode ? builder.setProducerLocation(location).buildLocal() : builder.buildRemote();
}
+
+ /**
+ * A {@link SingleInputGateFactory} which replaces the default {@link RemoteInputChannel} and
+ * {@link LocalInputChannel} implementation with costume ones.
+ */
+ private static class TestSingleInputGateFactory extends SingleInputGateFactory {
+
+ private final ResourceID taskExecutorResourceId;
Review comment:
Make all these fields as protected in parent `SingleInputGateFactory` for simple.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:FAILURE URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359 TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:PENDING URL:https://travis-ci.com/flink-ci/flink/builds/149795742 TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
-->
## CI report:
* fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Travis: [PENDING](https://travis-ci.com/flink-ci/flink/builds/149795742) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-590170854
Thanks for the updates @wsry and it generally looks good to me now, just left two tiny comments.
As for the previous main concern, I left some suggestions [here](https://github.com/apache/flink/pull/11155#discussion_r383088089).
After you providing the micro-benchmark results, I would also execute it in our benchmark machine to convince us.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] wsry commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
wsry commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r382414363
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -272,11 +287,161 @@ private static ShuffleDescriptor createShuffleDescriptor(
ResultPartitionID resultPartitionID,
ResourceID location,
TaskManagerLocation senderLocation,
- int channel) {
+ int connectionIndex) {
final NettyShuffleDescriptorBuilder builder = NettyShuffleDescriptorBuilder.newBuilder()
.setId(resultPartitionID)
.setProducerInfoFromTaskManagerLocation(senderLocation)
- .setConnectionIndex(channel);
+ .setConnectionIndex(connectionIndex);
return localMode ? builder.setProducerLocation(location).buildLocal() : builder.buildRemote();
}
+
+ /**
+ * A {@link SingleInputGateFactory} which replaces the default {@link RemoteInputChannel} and
+ * {@link LocalInputChannel} implementation with costume ones.
+ */
+ private static class TestSingleInputGateFactory extends SingleInputGateFactory {
+
+ private final ResourceID taskExecutorResourceId;
+ private final int partitionRequestInitialBackoff;
+ private final int partitionRequestMaxBackoff;
+ private final ConnectionManager connectionManager;
+ private final ResultPartitionManager partitionManager;
+ private final TaskEventPublisher taskEventPublisher;
+ private final NetworkBufferPool networkBufferPool;
+
+ public TestSingleInputGateFactory(
+ @Nonnull ResourceID taskExecutorResourceId,
+ @Nonnull NettyShuffleEnvironmentConfiguration networkConfig,
+ @Nonnull ConnectionManager connectionManager,
+ @Nonnull ResultPartitionManager partitionManager,
+ @Nonnull TaskEventPublisher taskEventPublisher,
+ @Nonnull NetworkBufferPool networkBufferPool) {
+ super(
+ taskExecutorResourceId,
+ networkConfig,
+ connectionManager,
+ partitionManager,
+ taskEventPublisher,
+ networkBufferPool);
+ this.networkBufferPool = networkBufferPool;
+ this.taskEventPublisher = taskEventPublisher;
+ this.partitionManager = partitionManager;
+ this.connectionManager = connectionManager;
+ this.partitionRequestMaxBackoff = networkConfig.partitionRequestMaxBackoff();
+ this.partitionRequestInitialBackoff = networkConfig.partitionRequestInitialBackoff();
+ this.taskExecutorResourceId = taskExecutorResourceId;
+ }
+
+ @Override
+ protected InputChannel createKnownInputChannel(
+ SingleInputGate inputGate,
+ int index,
+ NettyShuffleDescriptor inputChannelDescriptor,
+ ChannelStatistics channelStatistics,
+ InputChannelMetrics metrics) {
+ ResultPartitionID partitionId = inputChannelDescriptor.getResultPartitionID();
+ if (inputChannelDescriptor.isLocalTo(taskExecutorResourceId)) {
+ return new TestLocalInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics);
+ } else {
+ return new TestRemoteInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ inputChannelDescriptor.getConnectionId(),
+ connectionManager,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics,
+ networkBufferPool);
+ }
+ }
+ }
+
+ /**
+ * A {@link LocalInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestLocalInputChannel extends LocalInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
+
+ public TestLocalInputChannel(
+ SingleInputGate inputGate,
+ int channelIndex,
+ ResultPartitionID partitionId,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ int initialBackoff,
+ int maxBackoff,
+ InputChannelMetrics metrics) {
+ super(
+ inputGate,
+ channelIndex,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ initialBackoff,
+ maxBackoff,
+ metrics);
+ }
+
+ @Override
+ public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
Review comment:
Can we always call ```createResultPartitionWriter``` before call ```createReceiver```, which means the result partitions are always registered before requested. Current implementation can guarantee that. I am saying this because I need to override more than one method and also need do some refactor to support ```retriggerSubpartitionRequest```.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r382441933
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
##########
@@ -95,7 +95,7 @@ public LocalInputChannel(
// ------------------------------------------------------------------------
Review comment:
Please copy PR description into the commit message and once you have some expected benchmark results, please include them in the commit message as well :)
Could you also explain in both places, how are you solving the problem? Maybe something along those lines:
> Before this change, in network benchmark (for example 1000 channels benchmark with 4 record writers) StreamNetworkBenchmarkEnvironment#createInputGate was creating 1000 input gates with 4 input channels each, which doesn't make much sense. This commit is changing that to a single receiver with 4 input gates, with 1000 channels each.
>
> It is achieved by providing testing implementations of InputChannels, which are using channel index for requesting subpartitions from ResultPartition, instead of subpartition index. Thanks to that, we can connect single ResultPartition with N subpartitions, to a single instance of InputGate with 1000 channels.
?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149795742",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149835681",
"triggerID" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"triggerType" : "PUSH"
}, {
"hash" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149842861",
"triggerID" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149851288",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "090a72bd3101d7111490832526937d3973c75795",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149946832",
"triggerID" : "090a72bd3101d7111490832526937d3973c75795",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149949398",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5400",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/150266416",
"triggerID" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"triggerType" : "PUSH"
}, {
"hash" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5506",
"triggerID" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"triggerType" : "PUSH"
}, {
"hash" : "fbf92243b17619df704bf2600d294cda449c7ddc",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/flink-ci/flink/builds/150577037",
"triggerID" : "fbf92243b17619df704bf2600d294cda449c7ddc",
"triggerType" : "PUSH"
}, {
"hash" : "035041ed96b561f0d7bb69fa09436a24e42a5a54",
"status" : "PENDING",
"url" : "https://travis-ci.com/flink-ci/flink/builds/150774056",
"triggerID" : "035041ed96b561f0d7bb69fa09436a24e42a5a54",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* fbf92243b17619df704bf2600d294cda449c7ddc Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/150577037)
* 035041ed96b561f0d7bb69fa09436a24e42a5a54 Travis: [PENDING](https://travis-ci.com/flink-ci/flink/builds/150774056)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:FAILURE URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359 TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:FAILURE URL:https://travis-ci.com/flink-ci/flink/builds/149795742 TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
Hash:2d7b20f97e41151ed741a3f6d93756b776b30dc9 Status:CANCELED URL:https://travis-ci.com/flink-ci/flink/builds/149835681 TriggerType:PUSH TriggerID:2d7b20f97e41151ed741a3f6d93756b776b30dc9
Hash:cfc691c781ae54d63ecaa7c282e6cb34f73425d8 Status:PENDING URL:https://travis-ci.com/flink-ci/flink/builds/149842861 TriggerType:PUSH TriggerID:cfc691c781ae54d63ecaa7c282e6cb34f73425d8
Hash:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c Status:UNKNOWN URL:TBD TriggerType:PUSH TriggerID:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c
-->
## CI report:
* fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149795742) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359)
* 2d7b20f97e41151ed741a3f6d93756b776b30dc9 Travis: [CANCELED](https://travis-ci.com/flink-ci/flink/builds/149835681)
* cfc691c781ae54d63ecaa7c282e6cb34f73425d8 Travis: [PENDING](https://travis-ci.com/flink-ci/flink/builds/149842861)
* 21b36d4fadd73af33edb4fabc7da9d93c2f2d55c UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] wsry commented on issue #11155: [FLINK-14818][benchmark]
Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
wsry commented on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-591803930
@flinkbot run travis re-run the last Travis build
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r383090336
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/SingleInputGateBenchmarkFactory.java
##########
@@ -0,0 +1,173 @@
+/*
+ * 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.streaming.runtime.io.benchmark;
+
+import org.apache.flink.core.memory.MemorySegmentProvider;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.io.network.ConnectionID;
+import org.apache.flink.runtime.io.network.ConnectionManager;
+import org.apache.flink.runtime.io.network.TaskEventPublisher;
+import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+import org.apache.flink.runtime.io.network.metrics.InputChannelMetrics;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionManager;
+import org.apache.flink.runtime.io.network.partition.consumer.InputChannel;
+import org.apache.flink.runtime.io.network.partition.consumer.LocalInputChannel;
+import org.apache.flink.runtime.io.network.partition.consumer.RemoteInputChannel;
+import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate;
+import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGateFactory;
+import org.apache.flink.runtime.shuffle.NettyShuffleDescriptor;
+import org.apache.flink.runtime.taskmanager.NettyShuffleEnvironmentConfiguration;
+
+import java.io.IOException;
+
+/**
+ * A benchmark-specific input gate factory which overrides the respective methods of creating
+ * {@link RemoteInputChannel} and {@link LocalInputChannel} for requesting specific subpartitions.
+ */
+public class SingleInputGateBenchmarkFactory extends SingleInputGateFactory {
+
+ public SingleInputGateBenchmarkFactory(
+ ResourceID taskExecutorResourceId,
+ NettyShuffleEnvironmentConfiguration networkConfig,
+ ConnectionManager connectionManager,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ NetworkBufferPool networkBufferPool) {
+ super(
+ taskExecutorResourceId,
+ networkConfig,
+ connectionManager,
+ partitionManager,
+ taskEventPublisher,
+ networkBufferPool);
+ }
+
+ @Override
+ protected InputChannel createKnownInputChannel(
+ SingleInputGate inputGate,
+ int index,
+ NettyShuffleDescriptor inputChannelDescriptor,
+ SingleInputGateFactory.ChannelStatistics channelStatistics,
+ InputChannelMetrics metrics) {
+ ResultPartitionID partitionId = inputChannelDescriptor.getResultPartitionID();
+ if (inputChannelDescriptor.isLocalTo(taskExecutorResourceId)) {
+ return new TestLocalInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics);
+ } else {
+ return new TestRemoteInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ inputChannelDescriptor.getConnectionId(),
+ connectionManager,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics,
+ networkBufferPool);
+ }
+ }
+
+ /**
+ * A {@link LocalInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestLocalInputChannel extends LocalInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
+
+ public TestLocalInputChannel(
+ SingleInputGate inputGate,
+ int channelIndex,
+ ResultPartitionID partitionId,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ int initialBackoff,
+ int maxBackoff,
+ InputChannelMetrics metrics) {
+ super(
+ inputGate,
+ channelIndex,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ initialBackoff,
+ maxBackoff,
+ metrics);
+ }
+
+ @Override
+ public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ super.requestSubpartition(channelIndex);
+ }
+
+ @Override
+ public ResultPartitionID getPartitionId() {
+ return newPartitionID;
+ }
+ }
+
+ /**
+ * A {@link RemoteInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestRemoteInputChannel extends RemoteInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
+
+ public TestRemoteInputChannel(
+ SingleInputGate inputGate,
+ int channelIndex,
+ ResultPartitionID partitionId,
+ ConnectionID connectionId,
+ ConnectionManager connectionManager,
+ int initialBackOff,
+ int maxBackoff,
+ InputChannelMetrics metrics,
+ MemorySegmentProvider memorySegmentProvider) {
+ super(
+ inputGate,
+ channelIndex,
+ partitionId,
+ connectionId,
+ connectionManager,
+ initialBackOff,
+ maxBackoff,
+ metrics,
+ memorySegmentProvider);
+ }
+
+ @Override
+ public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ super.requestSubpartition(channelIndex);
+ }
+
+ @Override
+ public ResultPartitionID getPartitionId() {
Review comment:
Ignore this because of [comment](https://github.com/apache/flink/pull/11155#discussion_r383090260)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149795742",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149835681",
"triggerID" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"triggerType" : "PUSH"
}, {
"hash" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149842861",
"triggerID" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149851288",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "090a72bd3101d7111490832526937d3973c75795",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149946832",
"triggerID" : "090a72bd3101d7111490832526937d3973c75795",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149949398",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5400",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/150266416",
"triggerID" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"triggerType" : "PUSH"
}, {
"hash" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5506",
"triggerID" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"triggerType" : "PUSH"
}, {
"hash" : "fbf92243b17619df704bf2600d294cda449c7ddc",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/flink-ci/flink/builds/150577037",
"triggerID" : "fbf92243b17619df704bf2600d294cda449c7ddc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* fbf92243b17619df704bf2600d294cda449c7ddc Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/150577037)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r382397221
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -272,11 +287,161 @@ private static ShuffleDescriptor createShuffleDescriptor(
ResultPartitionID resultPartitionID,
ResourceID location,
TaskManagerLocation senderLocation,
- int channel) {
+ int connectionIndex) {
final NettyShuffleDescriptorBuilder builder = NettyShuffleDescriptorBuilder.newBuilder()
.setId(resultPartitionID)
.setProducerInfoFromTaskManagerLocation(senderLocation)
- .setConnectionIndex(channel);
+ .setConnectionIndex(connectionIndex);
return localMode ? builder.setProducerLocation(location).buildLocal() : builder.buildRemote();
}
+
+ /**
+ * A {@link SingleInputGateFactory} which replaces the default {@link RemoteInputChannel} and
Review comment:
We should adjust the descriptions a bit. typo `costume`
E.g. A benchmark-specific input gate factory which overrides the respective methods of creating {@link RemotInputChannel} and {@link LocalInputChannel} for requesting specific subpartition.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] wsry commented on issue #11155: [FLINK-14818][benchmark]
Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
wsry commented on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-590255760
Thanks for the review and comments. @zhijiangW @pnowojski
I have updated the PR according to the comments and the benchmark results are also attached.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149795742",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149835681",
"triggerID" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"triggerType" : "PUSH"
}, {
"hash" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149842861",
"triggerID" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149851288",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "090a72bd3101d7111490832526937d3973c75795",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149946832",
"triggerID" : "090a72bd3101d7111490832526937d3973c75795",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149949398",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5400",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/flink-ci/flink/builds/150266416",
"triggerID" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"triggerType" : "PUSH"
}, {
"hash" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5506",
"triggerID" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"triggerType" : "PUSH"
}, {
"hash" : "fbf92243b17619df704bf2600d294cda449c7ddc",
"status" : "PENDING",
"url" : "https://travis-ci.com/flink-ci/flink/builds/150577037",
"triggerID" : "fbf92243b17619df704bf2600d294cda449c7ddc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8f263fe38efaf89c0f001d3995d64e7bd46493b1 Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/150266416) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5506)
* fbf92243b17619df704bf2600d294cda449c7ddc Travis: [PENDING](https://travis-ci.com/flink-ci/flink/builds/150577037)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-590753152
Also attach the comparison results executed in benchmark machine:
Baseline | |
-- | -- | --
DataSkewStreamNetworkThroughputBenchmarkExecutor| 51977.13475 |
StreamNetworkBroadcastThroughputBenchmarkExecutor| 1164.463375 |
StreamNetworkThroughputBenchmarkExecutor| 69553.22069 | 100,100ms
StreamNetworkThroughputBenchmarkExecutor| 17430.11747 | 100,100ms,SSL
StreamNetworkThroughputBenchmarkExecutor| 51019.51607 | 1000,1ms
StreamNetworkThroughputBenchmarkExecutor| 55423.23278 | 1000,100ms
StreamNetworkThroughputBenchmarkExecutor| 14924.65899 | 1000,100ms,SSL
PR change | |
DataSkewStreamNetworkThroughputBenchmarkExecutor| 33215.6767 |
StreamNetworkBroadcastThroughputBenchmarkExecutor| 1189.239694 |
StreamNetworkThroughputBenchmarkExecutor| 72450.69576 | 100,100ms
StreamNetworkThroughputBenchmarkExecutor| 17018.46785 | 100,100ms,SSL
StreamNetworkThroughputBenchmarkExecutor| 45829.91493 | 1000,1ms
StreamNetworkThroughputBenchmarkExecutor| 52943.20546 | 1000,100ms
StreamNetworkThroughputBenchmarkExecutor| 13762.96631 | 1000,100ms,SSL
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r382389121
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -272,11 +287,161 @@ private static ShuffleDescriptor createShuffleDescriptor(
ResultPartitionID resultPartitionID,
ResourceID location,
TaskManagerLocation senderLocation,
- int channel) {
+ int connectionIndex) {
final NettyShuffleDescriptorBuilder builder = NettyShuffleDescriptorBuilder.newBuilder()
.setId(resultPartitionID)
.setProducerInfoFromTaskManagerLocation(senderLocation)
- .setConnectionIndex(channel);
+ .setConnectionIndex(connectionIndex);
return localMode ? builder.setProducerLocation(location).buildLocal() : builder.buildRemote();
}
+
+ /**
+ * A {@link SingleInputGateFactory} which replaces the default {@link RemoteInputChannel} and
+ * {@link LocalInputChannel} implementation with costume ones.
+ */
+ private static class TestSingleInputGateFactory extends SingleInputGateFactory {
+
+ private final ResourceID taskExecutorResourceId;
+ private final int partitionRequestInitialBackoff;
+ private final int partitionRequestMaxBackoff;
+ private final ConnectionManager connectionManager;
+ private final ResultPartitionManager partitionManager;
+ private final TaskEventPublisher taskEventPublisher;
+ private final NetworkBufferPool networkBufferPool;
+
+ public TestSingleInputGateFactory(
+ @Nonnull ResourceID taskExecutorResourceId,
+ @Nonnull NettyShuffleEnvironmentConfiguration networkConfig,
+ @Nonnull ConnectionManager connectionManager,
+ @Nonnull ResultPartitionManager partitionManager,
+ @Nonnull TaskEventPublisher taskEventPublisher,
+ @Nonnull NetworkBufferPool networkBufferPool) {
+ super(
+ taskExecutorResourceId,
+ networkConfig,
+ connectionManager,
+ partitionManager,
+ taskEventPublisher,
+ networkBufferPool);
+ this.networkBufferPool = networkBufferPool;
+ this.taskEventPublisher = taskEventPublisher;
+ this.partitionManager = partitionManager;
+ this.connectionManager = connectionManager;
+ this.partitionRequestMaxBackoff = networkConfig.partitionRequestMaxBackoff();
+ this.partitionRequestInitialBackoff = networkConfig.partitionRequestInitialBackoff();
+ this.taskExecutorResourceId = taskExecutorResourceId;
+ }
+
+ @Override
+ protected InputChannel createKnownInputChannel(
+ SingleInputGate inputGate,
Review comment:
nit: one more indentation for distinguishing with the below core codes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r382445336
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -238,18 +241,19 @@ private InputGate createInputGate(TaskManagerLocation senderLocation) throws Exc
private InputGateDeploymentDescriptor createInputGateDeploymentDescriptor(
TaskManagerLocation senderLocation,
- int consumedSubpartitionIndex,
+ int gateIndex,
ResourceID localLocation) {
- final ShuffleDescriptor[] channelDescriptors = Arrays.stream(partitionIds)
- .map(partitionId ->
- createShuffleDescriptor(localMode, partitionId, localLocation, senderLocation, consumedSubpartitionIndex))
- .toArray(ShuffleDescriptor[]::new);
+ final ShuffleDescriptor[] channelDescriptors = new ShuffleDescriptor[channels];
+ for (int channelIndex = 0; channelIndex < channels; ++channelIndex) {
+ channelDescriptors[channelIndex] = createShuffleDescriptor(
+ localMode, partitionIds[gateIndex], localLocation, senderLocation, gateIndex);
+ }
return new InputGateDeploymentDescriptor(
dataSetID,
ResultPartitionType.PIPELINED_BOUNDED,
- consumedSubpartitionIndex,
+ 0,
Review comment:
This is always zero, as it's being ignored by `Test***InputChannel` classes, right?
If so, maybe add a comment explaining that?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-591277621
@wsry I also updated the benchmark results with the same connection amount in [link](https://github.com/apache/flink/pull/11155#issuecomment-590753152). The conclusion is that all the cases for network throughput have regression except for 100, 100ms. I am not sure whether we can explain this corner case properly. But I guess it would not block merging since it should be the right way to go.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:FAILURE URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359 TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:FAILURE URL:https://travis-ci.com/flink-ci/flink/builds/149795742 TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
Hash:2d7b20f97e41151ed741a3f6d93756b776b30dc9 Status:CANCELED URL:https://travis-ci.com/flink-ci/flink/builds/149835681 TriggerType:PUSH TriggerID:2d7b20f97e41151ed741a3f6d93756b776b30dc9
Hash:cfc691c781ae54d63ecaa7c282e6cb34f73425d8 Status:CANCELED URL:https://travis-ci.com/flink-ci/flink/builds/149842861 TriggerType:PUSH TriggerID:cfc691c781ae54d63ecaa7c282e6cb34f73425d8
Hash:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c Status:SUCCESS URL:https://travis-ci.com/flink-ci/flink/builds/149851288 TriggerType:PUSH TriggerID:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c
Hash:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c Status:SUCCESS URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381 TriggerType:PUSH TriggerID:21b36d4fadd73af33edb4fabc7da9d93c2f2d55c
Hash:090a72bd3101d7111490832526937d3973c75795 Status:FAILURE URL:https://travis-ci.com/flink-ci/flink/builds/149946832 TriggerType:PUSH TriggerID:090a72bd3101d7111490832526937d3973c75795
Hash:8fe5e3d33d372b15e15cc52171ea8c3844c02ef9 Status:SUCCESS URL:https://travis-ci.com/flink-ci/flink/builds/149949398 TriggerType:PUSH TriggerID:8fe5e3d33d372b15e15cc52171ea8c3844c02ef9
Hash:8fe5e3d33d372b15e15cc52171ea8c3844c02ef9 Status:SUCCESS URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5400 TriggerType:PUSH TriggerID:8fe5e3d33d372b15e15cc52171ea8c3844c02ef9
-->
## CI report:
* fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149795742) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359)
* 2d7b20f97e41151ed741a3f6d93756b776b30dc9 Travis: [CANCELED](https://travis-ci.com/flink-ci/flink/builds/149835681)
* cfc691c781ae54d63ecaa7c282e6cb34f73425d8 Travis: [CANCELED](https://travis-ci.com/flink-ci/flink/builds/149842861)
* 21b36d4fadd73af33edb4fabc7da9d93c2f2d55c Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/149851288) Azure: [SUCCESS](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381)
* 090a72bd3101d7111490832526937d3973c75795 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149946832)
* 8fe5e3d33d372b15e15cc52171ea8c3844c02ef9 Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/149949398) Azure: [SUCCESS](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5400)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r382389476
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -272,11 +287,161 @@ private static ShuffleDescriptor createShuffleDescriptor(
ResultPartitionID resultPartitionID,
ResourceID location,
TaskManagerLocation senderLocation,
- int channel) {
+ int connectionIndex) {
final NettyShuffleDescriptorBuilder builder = NettyShuffleDescriptorBuilder.newBuilder()
.setId(resultPartitionID)
.setProducerInfoFromTaskManagerLocation(senderLocation)
- .setConnectionIndex(channel);
+ .setConnectionIndex(connectionIndex);
return localMode ? builder.setProducerLocation(location).buildLocal() : builder.buildRemote();
}
+
+ /**
+ * A {@link SingleInputGateFactory} which replaces the default {@link RemoteInputChannel} and
+ * {@link LocalInputChannel} implementation with costume ones.
+ */
+ private static class TestSingleInputGateFactory extends SingleInputGateFactory {
Review comment:
nit: We can make it a separate class to void this class too long. Also rename it `SingleInputGateBenchmarkFactory`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-590753152
Also attach the comparison results executed in benchmark machine:
Baseline | |
-- | -- | --
DataSkewStreamNetworkThroughputBenchmarkExecutor| 51977.13475 |
StreamNetworkBroadcastThroughputBenchmarkExecutor| 1164.463375 |
StreamNetworkThroughputBenchmarkExecutor| 69553.22069 | 100,100ms
StreamNetworkThroughputBenchmarkExecutor| 17430.11747 | 100,100ms,SSL
StreamNetworkThroughputBenchmarkExecutor| 51019.51607 | 1000,1ms
StreamNetworkThroughputBenchmarkExecutor| 55423.23278 | 1000,100ms
StreamNetworkThroughputBenchmarkExecutor| 14924.65899 | 1000,100ms,SSL
less gates with less connections | |
DataSkewStreamNetworkThroughputBenchmarkExecutor| 33215.6767 |
StreamNetworkBroadcastThroughputBenchmarkExecutor| 1189.239694 |
StreamNetworkThroughputBenchmarkExecutor| 72450.69576 | 100,100ms
StreamNetworkThroughputBenchmarkExecutor| 17018.46785 | 100,100ms,SSL
StreamNetworkThroughputBenchmarkExecutor| 45829.91493 | 1000,1ms
StreamNetworkThroughputBenchmarkExecutor| 52943.20546 | 1000,100ms
StreamNetworkThroughputBenchmarkExecutor| 13762.96631 | 1000,100ms,SSL
less gates with same connections | |
DataSkewStreamNetworkThroughputBenchmarkExecutor| 32342.12887|
StreamNetworkBroadcastThroughputBenchmarkExecutor| 1136.44912 |
StreamNetworkThroughputBenchmarkExecutor| 75151.951622 | 100,100ms
StreamNetworkThroughputBenchmarkExecutor| 16878.544466 | 100,100ms,SSL
StreamNetworkThroughputBenchmarkExecutor| 47041.479598 | 1000,1ms
StreamNetworkThroughputBenchmarkExecutor| 52384.517795 | 1000,100ms
StreamNetworkThroughputBenchmarkExecutor| 13824.809088 | 1000,100ms,SSL
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:FAILURE URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359 TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
Hash:fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Status:FAILURE URL:https://travis-ci.com/flink-ci/flink/builds/149795742 TriggerType:PUSH TriggerID:fee93cecf80cb73fcdf082f47fe0b9bd727a3002
-->
## CI report:
* fee93cecf80cb73fcdf082f47fe0b9bd727a3002 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149795742) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r382398126
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
##########
@@ -95,7 +96,8 @@ public LocalInputChannel(
// ------------------------------------------------------------------------
@Override
- void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ @VisibleForTesting
Review comment:
I guess it is not necessary to add this annotation which only indicates for `private` change.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r382390727
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -272,11 +287,161 @@ private static ShuffleDescriptor createShuffleDescriptor(
ResultPartitionID resultPartitionID,
ResourceID location,
TaskManagerLocation senderLocation,
- int channel) {
+ int connectionIndex) {
final NettyShuffleDescriptorBuilder builder = NettyShuffleDescriptorBuilder.newBuilder()
.setId(resultPartitionID)
.setProducerInfoFromTaskManagerLocation(senderLocation)
- .setConnectionIndex(channel);
+ .setConnectionIndex(connectionIndex);
return localMode ? builder.setProducerLocation(location).buildLocal() : builder.buildRemote();
}
+
+ /**
+ * A {@link SingleInputGateFactory} which replaces the default {@link RemoteInputChannel} and
+ * {@link LocalInputChannel} implementation with costume ones.
+ */
+ private static class TestSingleInputGateFactory extends SingleInputGateFactory {
+
+ private final ResourceID taskExecutorResourceId;
+ private final int partitionRequestInitialBackoff;
+ private final int partitionRequestMaxBackoff;
+ private final ConnectionManager connectionManager;
+ private final ResultPartitionManager partitionManager;
+ private final TaskEventPublisher taskEventPublisher;
+ private final NetworkBufferPool networkBufferPool;
+
+ public TestSingleInputGateFactory(
+ @Nonnull ResourceID taskExecutorResourceId,
+ @Nonnull NettyShuffleEnvironmentConfiguration networkConfig,
+ @Nonnull ConnectionManager connectionManager,
+ @Nonnull ResultPartitionManager partitionManager,
+ @Nonnull TaskEventPublisher taskEventPublisher,
+ @Nonnull NetworkBufferPool networkBufferPool) {
+ super(
+ taskExecutorResourceId,
+ networkConfig,
+ connectionManager,
+ partitionManager,
+ taskEventPublisher,
+ networkBufferPool);
+ this.networkBufferPool = networkBufferPool;
+ this.taskEventPublisher = taskEventPublisher;
+ this.partitionManager = partitionManager;
+ this.connectionManager = connectionManager;
+ this.partitionRequestMaxBackoff = networkConfig.partitionRequestMaxBackoff();
+ this.partitionRequestInitialBackoff = networkConfig.partitionRequestInitialBackoff();
+ this.taskExecutorResourceId = taskExecutorResourceId;
+ }
+
+ @Override
+ protected InputChannel createKnownInputChannel(
+ SingleInputGate inputGate,
+ int index,
+ NettyShuffleDescriptor inputChannelDescriptor,
+ ChannelStatistics channelStatistics,
+ InputChannelMetrics metrics) {
+ ResultPartitionID partitionId = inputChannelDescriptor.getResultPartitionID();
+ if (inputChannelDescriptor.isLocalTo(taskExecutorResourceId)) {
+ return new TestLocalInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics);
+ } else {
+ return new TestRemoteInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ inputChannelDescriptor.getConnectionId(),
+ connectionManager,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics,
+ networkBufferPool);
+ }
+ }
+ }
+
+ /**
+ * A {@link LocalInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestLocalInputChannel extends LocalInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
+
+ public TestLocalInputChannel(
+ SingleInputGate inputGate,
+ int channelIndex,
+ ResultPartitionID partitionId,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ int initialBackoff,
+ int maxBackoff,
+ InputChannelMetrics metrics) {
+ super(
+ inputGate,
+ channelIndex,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ initialBackoff,
+ maxBackoff,
+ metrics);
+ }
+
+ @Override
+ public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
Review comment:
In addition, we might also need to override the method of `retriggerSubpartitionRequest`, in case this process might be used to cause invalid `subpartitionIndex` future.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r383081286
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -158,6 +157,10 @@ public void tearDown() {
suppressExceptions(receiverEnv::close);
}
+ /**
+ * Note: It should be guaranteed that {@link #createResultPartitionWriter(int)} has been called before
+ * creating the receiver.
Review comment:
nit: might supplement the reason why we should guarantee this.
E.g. `Otherwise it might cause unexpected behaviors when {@link PartitionNotFoundException} happens in {@link TestRemoteInputChannel}`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r383102059
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -272,11 +287,161 @@ private static ShuffleDescriptor createShuffleDescriptor(
ResultPartitionID resultPartitionID,
ResourceID location,
TaskManagerLocation senderLocation,
- int channel) {
+ int connectionIndex) {
final NettyShuffleDescriptorBuilder builder = NettyShuffleDescriptorBuilder.newBuilder()
.setId(resultPartitionID)
.setProducerInfoFromTaskManagerLocation(senderLocation)
- .setConnectionIndex(channel);
+ .setConnectionIndex(connectionIndex);
return localMode ? builder.setProducerLocation(location).buildLocal() : builder.buildRemote();
}
+
+ /**
+ * A {@link SingleInputGateFactory} which replaces the default {@link RemoteInputChannel} and
+ * {@link LocalInputChannel} implementation with costume ones.
+ */
+ private static class TestSingleInputGateFactory extends SingleInputGateFactory {
+
+ private final ResourceID taskExecutorResourceId;
+ private final int partitionRequestInitialBackoff;
+ private final int partitionRequestMaxBackoff;
+ private final ConnectionManager connectionManager;
+ private final ResultPartitionManager partitionManager;
+ private final TaskEventPublisher taskEventPublisher;
+ private final NetworkBufferPool networkBufferPool;
+
+ public TestSingleInputGateFactory(
+ @Nonnull ResourceID taskExecutorResourceId,
+ @Nonnull NettyShuffleEnvironmentConfiguration networkConfig,
+ @Nonnull ConnectionManager connectionManager,
+ @Nonnull ResultPartitionManager partitionManager,
+ @Nonnull TaskEventPublisher taskEventPublisher,
+ @Nonnull NetworkBufferPool networkBufferPool) {
+ super(
+ taskExecutorResourceId,
+ networkConfig,
+ connectionManager,
+ partitionManager,
+ taskEventPublisher,
+ networkBufferPool);
+ this.networkBufferPool = networkBufferPool;
+ this.taskEventPublisher = taskEventPublisher;
+ this.partitionManager = partitionManager;
+ this.connectionManager = connectionManager;
+ this.partitionRequestMaxBackoff = networkConfig.partitionRequestMaxBackoff();
+ this.partitionRequestInitialBackoff = networkConfig.partitionRequestInitialBackoff();
+ this.taskExecutorResourceId = taskExecutorResourceId;
+ }
+
+ @Override
+ protected InputChannel createKnownInputChannel(
+ SingleInputGate inputGate,
+ int index,
+ NettyShuffleDescriptor inputChannelDescriptor,
+ ChannelStatistics channelStatistics,
+ InputChannelMetrics metrics) {
+ ResultPartitionID partitionId = inputChannelDescriptor.getResultPartitionID();
+ if (inputChannelDescriptor.isLocalTo(taskExecutorResourceId)) {
+ return new TestLocalInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics);
+ } else {
+ return new TestRemoteInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ inputChannelDescriptor.getConnectionId(),
+ connectionManager,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics,
+ networkBufferPool);
+ }
+ }
+ }
+
+ /**
+ * A {@link LocalInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestLocalInputChannel extends LocalInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
+
+ public TestLocalInputChannel(
+ SingleInputGate inputGate,
+ int channelIndex,
+ ResultPartitionID partitionId,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ int initialBackoff,
+ int maxBackoff,
+ InputChannelMetrics metrics) {
+ super(
+ inputGate,
+ channelIndex,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ initialBackoff,
+ maxBackoff,
+ metrics);
+ }
+
+ @Override
+ public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ super.requestSubpartition(channelIndex);
+ }
+
+ @Override
+ public ResultPartitionID getPartitionId() {
Review comment:
I forgot the case of requesting partition via `ResultPartitionID`, so the above option for `ResultPartitionID` is invalid.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] wsry commented on issue #11155: [FLINK-14818][benchmark]
Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
wsry commented on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-589529462
I have updated the PR. @zhijiangW
I will attach the new benchmark results soon.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] wsry commented on issue #11155: [FLINK-14818][benchmark]
Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
wsry commented on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-591814004
@flinkbot run travis
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149795742",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149835681",
"triggerID" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"triggerType" : "PUSH"
}, {
"hash" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149842861",
"triggerID" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149851288",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "090a72bd3101d7111490832526937d3973c75795",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149946832",
"triggerID" : "090a72bd3101d7111490832526937d3973c75795",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149949398",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5400",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "07c2811b17d4ee4bb517b5b1a98d7a3bc91f2dea",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "07c2811b17d4ee4bb517b5b1a98d7a3bc91f2dea",
"triggerType" : "PUSH"
}, {
"hash" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"status" : "PENDING",
"url" : "https://travis-ci.com/flink-ci/flink/builds/150266416",
"triggerID" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8fe5e3d33d372b15e15cc52171ea8c3844c02ef9 Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/149949398) Azure: [SUCCESS](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5400)
* 07c2811b17d4ee4bb517b5b1a98d7a3bc91f2dea UNKNOWN
* 8f263fe38efaf89c0f001d3995d64e7bd46493b1 Travis: [PENDING](https://travis-ci.com/flink-ci/flink/builds/150266416)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r383090816
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -272,11 +287,161 @@ private static ShuffleDescriptor createShuffleDescriptor(
ResultPartitionID resultPartitionID,
ResourceID location,
TaskManagerLocation senderLocation,
- int channel) {
+ int connectionIndex) {
final NettyShuffleDescriptorBuilder builder = NettyShuffleDescriptorBuilder.newBuilder()
.setId(resultPartitionID)
.setProducerInfoFromTaskManagerLocation(senderLocation)
- .setConnectionIndex(channel);
+ .setConnectionIndex(connectionIndex);
return localMode ? builder.setProducerLocation(location).buildLocal() : builder.buildRemote();
}
+
+ /**
+ * A {@link SingleInputGateFactory} which replaces the default {@link RemoteInputChannel} and
+ * {@link LocalInputChannel} implementation with costume ones.
+ */
+ private static class TestSingleInputGateFactory extends SingleInputGateFactory {
+
+ private final ResourceID taskExecutorResourceId;
+ private final int partitionRequestInitialBackoff;
+ private final int partitionRequestMaxBackoff;
+ private final ConnectionManager connectionManager;
+ private final ResultPartitionManager partitionManager;
+ private final TaskEventPublisher taskEventPublisher;
+ private final NetworkBufferPool networkBufferPool;
+
+ public TestSingleInputGateFactory(
+ @Nonnull ResourceID taskExecutorResourceId,
+ @Nonnull NettyShuffleEnvironmentConfiguration networkConfig,
+ @Nonnull ConnectionManager connectionManager,
+ @Nonnull ResultPartitionManager partitionManager,
+ @Nonnull TaskEventPublisher taskEventPublisher,
+ @Nonnull NetworkBufferPool networkBufferPool) {
+ super(
+ taskExecutorResourceId,
+ networkConfig,
+ connectionManager,
+ partitionManager,
+ taskEventPublisher,
+ networkBufferPool);
+ this.networkBufferPool = networkBufferPool;
+ this.taskEventPublisher = taskEventPublisher;
+ this.partitionManager = partitionManager;
+ this.connectionManager = connectionManager;
+ this.partitionRequestMaxBackoff = networkConfig.partitionRequestMaxBackoff();
+ this.partitionRequestInitialBackoff = networkConfig.partitionRequestInitialBackoff();
+ this.taskExecutorResourceId = taskExecutorResourceId;
+ }
+
+ @Override
+ protected InputChannel createKnownInputChannel(
+ SingleInputGate inputGate,
+ int index,
+ NettyShuffleDescriptor inputChannelDescriptor,
+ ChannelStatistics channelStatistics,
+ InputChannelMetrics metrics) {
+ ResultPartitionID partitionId = inputChannelDescriptor.getResultPartitionID();
+ if (inputChannelDescriptor.isLocalTo(taskExecutorResourceId)) {
+ return new TestLocalInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics);
+ } else {
+ return new TestRemoteInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ inputChannelDescriptor.getConnectionId(),
+ connectionManager,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics,
+ networkBufferPool);
+ }
+ }
+ }
+
+ /**
+ * A {@link LocalInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestLocalInputChannel extends LocalInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
+
+ public TestLocalInputChannel(
+ SingleInputGate inputGate,
+ int channelIndex,
+ ResultPartitionID partitionId,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ int initialBackoff,
+ int maxBackoff,
+ InputChannelMetrics metrics) {
+ super(
+ inputGate,
+ channelIndex,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ initialBackoff,
+ maxBackoff,
+ metrics);
+ }
+
+ @Override
+ public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ super.requestSubpartition(channelIndex);
+ }
+
+ @Override
+ public ResultPartitionID getPartitionId() {
Review comment:
While reviewing the current usage of `InputChannel#getPartitionId`, I find it is mainly used for generating `AddCredit` message. But the `ResultPartitionID` was actually not used for upstream side in practice.
So I prefer to remove this field from `AddCredit` message, which can further reduce the message size from current 52 bytes to 20 bytes.
@pnowojski do you have other concerns for removing it? If not, i would create a separate ticket for this issue.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#discussion_r382443220
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
##########
@@ -272,11 +287,161 @@ private static ShuffleDescriptor createShuffleDescriptor(
ResultPartitionID resultPartitionID,
ResourceID location,
TaskManagerLocation senderLocation,
- int channel) {
+ int connectionIndex) {
final NettyShuffleDescriptorBuilder builder = NettyShuffleDescriptorBuilder.newBuilder()
.setId(resultPartitionID)
.setProducerInfoFromTaskManagerLocation(senderLocation)
- .setConnectionIndex(channel);
+ .setConnectionIndex(connectionIndex);
return localMode ? builder.setProducerLocation(location).buildLocal() : builder.buildRemote();
}
+
+ /**
+ * A {@link SingleInputGateFactory} which replaces the default {@link RemoteInputChannel} and
+ * {@link LocalInputChannel} implementation with costume ones.
+ */
+ private static class TestSingleInputGateFactory extends SingleInputGateFactory {
+
+ private final ResourceID taskExecutorResourceId;
+ private final int partitionRequestInitialBackoff;
+ private final int partitionRequestMaxBackoff;
+ private final ConnectionManager connectionManager;
+ private final ResultPartitionManager partitionManager;
+ private final TaskEventPublisher taskEventPublisher;
+ private final NetworkBufferPool networkBufferPool;
+
+ public TestSingleInputGateFactory(
+ @Nonnull ResourceID taskExecutorResourceId,
+ @Nonnull NettyShuffleEnvironmentConfiguration networkConfig,
+ @Nonnull ConnectionManager connectionManager,
+ @Nonnull ResultPartitionManager partitionManager,
+ @Nonnull TaskEventPublisher taskEventPublisher,
+ @Nonnull NetworkBufferPool networkBufferPool) {
+ super(
+ taskExecutorResourceId,
+ networkConfig,
+ connectionManager,
+ partitionManager,
+ taskEventPublisher,
+ networkBufferPool);
+ this.networkBufferPool = networkBufferPool;
+ this.taskEventPublisher = taskEventPublisher;
+ this.partitionManager = partitionManager;
+ this.connectionManager = connectionManager;
+ this.partitionRequestMaxBackoff = networkConfig.partitionRequestMaxBackoff();
+ this.partitionRequestInitialBackoff = networkConfig.partitionRequestInitialBackoff();
+ this.taskExecutorResourceId = taskExecutorResourceId;
+ }
+
+ @Override
+ protected InputChannel createKnownInputChannel(
+ SingleInputGate inputGate,
+ int index,
+ NettyShuffleDescriptor inputChannelDescriptor,
+ ChannelStatistics channelStatistics,
+ InputChannelMetrics metrics) {
+ ResultPartitionID partitionId = inputChannelDescriptor.getResultPartitionID();
+ if (inputChannelDescriptor.isLocalTo(taskExecutorResourceId)) {
+ return new TestLocalInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics);
+ } else {
+ return new TestRemoteInputChannel(
+ inputGate,
+ index,
+ partitionId,
+ inputChannelDescriptor.getConnectionId(),
+ connectionManager,
+ partitionRequestInitialBackoff,
+ partitionRequestMaxBackoff,
+ metrics,
+ networkBufferPool);
+ }
+ }
+ }
+
+ /**
+ * A {@link LocalInputChannel} which ignores the given subpartition index and uses channel index
+ * instead when requesting subpartition.
+ */
+ private static class TestLocalInputChannel extends LocalInputChannel {
+
+ private ResultPartitionID newPartitionID = new ResultPartitionID();
+
+ public TestLocalInputChannel(
+ SingleInputGate inputGate,
+ int channelIndex,
+ ResultPartitionID partitionId,
+ ResultPartitionManager partitionManager,
+ TaskEventPublisher taskEventPublisher,
+ int initialBackoff,
+ int maxBackoff,
+ InputChannelMetrics metrics) {
+ super(
+ inputGate,
+ channelIndex,
+ partitionId,
+ partitionManager,
+ taskEventPublisher,
+ initialBackoff,
+ maxBackoff,
+ metrics);
+ }
+
+ @Override
+ public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ super.requestSubpartition(channelIndex);
+ }
+
+ @Override
+ public ResultPartitionID getPartitionId() {
Review comment:
I get the point and it makes me a bit torn whether this way is reasonable to go.
We hacked two internal infos (`ResultPartitionID` and `subpartitionIndex`) of `SingleInputGate` and `InputChannel`. I wonder it might make the benchmark very fragile in future. I mean if there are any changes related to these infos inside gate and channel, then it might bring unexpected behaviors for benchmark, because the internal logics would still access the previous inconsistent infos.
I need to think it through then.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149795742",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149835681",
"triggerID" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"triggerType" : "PUSH"
}, {
"hash" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149842861",
"triggerID" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149851288",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "090a72bd3101d7111490832526937d3973c75795",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149946832",
"triggerID" : "090a72bd3101d7111490832526937d3973c75795",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149949398",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5400",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/flink-ci/flink/builds/150266416",
"triggerID" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"triggerType" : "PUSH"
}, {
"hash" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5506",
"triggerID" : "8f263fe38efaf89c0f001d3995d64e7bd46493b1",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8f263fe38efaf89c0f001d3995d64e7bd46493b1 Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/150266416) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5506)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11155:
[FLINK-14818][benchmark] Fix receiving InputGate setup of
StreamNetworkBenchmarkEnvironment.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11155: [FLINK-14818][benchmark] Fix receiving InputGate setup of StreamNetworkBenchmarkEnvironment.
URL: https://github.com/apache/flink/pull/11155#issuecomment-588965557
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5359",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149795742",
"triggerID" : "fee93cecf80cb73fcdf082f47fe0b9bd727a3002",
"triggerType" : "PUSH"
}, {
"hash" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149835681",
"triggerID" : "2d7b20f97e41151ed741a3f6d93756b776b30dc9",
"triggerType" : "PUSH"
}, {
"hash" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149842861",
"triggerID" : "cfc691c781ae54d63ecaa7c282e6cb34f73425d8",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149851288",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5381",
"triggerID" : "21b36d4fadd73af33edb4fabc7da9d93c2f2d55c",
"triggerType" : "PUSH"
}, {
"hash" : "090a72bd3101d7111490832526937d3973c75795",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149946832",
"triggerID" : "090a72bd3101d7111490832526937d3973c75795",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149949398",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5400",
"triggerID" : "8fe5e3d33d372b15e15cc52171ea8c3844c02ef9",
"triggerType" : "PUSH"
}, {
"hash" : "07c2811b17d4ee4bb517b5b1a98d7a3bc91f2dea",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "07c2811b17d4ee4bb517b5b1a98d7a3bc91f2dea",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8fe5e3d33d372b15e15cc52171ea8c3844c02ef9 Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/149949398) Azure: [SUCCESS](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5400)
* 07c2811b17d4ee4bb517b5b1a98d7a3bc91f2dea UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services