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:
us...@infra.apache.org
With regards,
Apache Git Services
