[FLINK-12331] Introduce partition/gate setup to decouple task registration with NetworkEnvironment #8416

azagrebin · 2019-05-10T20:20:35Z

What is the purpose of the change

Introduce partition/gate setup to decouple task registration with NetworkEnvironment.
The PR also does some preparation refactoring.

Brief change log

Introduce factories for ResultPartion and SingleInputgates
Introduce buffer pool factories for ResultPartion and SingleInputgates
Introduce MemorySegmentProvider for RemoteInputChannel to assign exclusive segments
Refactor NetworkEnvironment#setupXXX() to ResultPartitionWriter#setup() and InputGate#setup()

Verifying this change

existing unit tests.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (no)
If yes, how is the feature documented? (not applicable)

flinkbot · 2019-05-10T20:21:57Z

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.

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❗ 3. Needs [attention] from.
- Needs attention by @zhijiangW
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

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

Bot commands

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

azagrebin · 2019-05-10T20:22:41Z

@flinkbot attention @zhijiangW

zhijiangW · 2019-05-13T03:46:40Z

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java

@@ -108,6 +112,7 @@ private NetworkEnvironment(
 		this.taskEventPublisher = taskEventPublisher;
 		this.isShutdown = false;


Put isShutdown=false at the end of this constructor?

zhijiangW · 2019-05-13T03:48:42Z

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java

@@ -128,13 +133,17 @@ public static NetworkEnvironment create(
 		ResultPartitionFactory resultPartitionFactory =
 			new ResultPartitionFactory(resultPartitionManager, checkNotNull(ioManager));

+		SingleInputGateFactory singleInputGateFactory =
+			new SingleInputGateFactory(config, connectionManager, resultPartitionManager);


TaskEventDispatcher could also be included in singleInputGateFactory?

zhijiangW · 2019-05-13T04:15:10Z

...rc/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java

+				InputChannelBuilder.newBuilder()
+					.setupFromNetworkEnvironment(network)
+					.buildUnknown(inputGate)
+					.toRemoteInputChannel(InputChannelBuilder.STUB_CONNECTION_ID);


why not build remote directly?

I tried to preserve the previous way of building remote but looking at it more, true, it does not seem to be relevant.

zhijiangW · 2019-05-13T04:15:42Z

...rc/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java

+					.buildUnknown(inputGate)
+					.toRemoteInputChannel(InputChannelBuilder.STUB_CONNECTION_ID);
+			IntermediateResultPartitionID resultPartitionId = remoteFromUnknown.getPartitionId().getPartitionId();
+			inputGate.setInputChannel(resultPartitionId, remoteFromUnknown);


setInputChannel could be removed?

zhijiangW · 2019-05-13T04:17:06Z

...rc/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java


 			network.setupInputGate(inputGate);

 			NetworkBufferPool bufferPool = network.getNetworkBufferPool();
 			if (enableCreditBasedFlowControl) {
-				RemoteInputChannel remote = (RemoteInputChannel) inputGate.getInputChannels()
-					.get(resultPartitionId.getPartitionId());
+				RemoteInputChannel remote = (RemoteInputChannel) inputGate.getInputChannels().get(resultPartitionId);


Use the previous remoteFromUnknown directly, so IntermediateResultPartitionID resultPartitionId = remoteFromUnknown.getPartitionId().getPartitionId() could also be removed.

zhijiangW · 2019-05-13T06:08:57Z

flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentProvider.java

+	/** Requests default number of memory segments. */
+	Collection<MemorySegment> requestMemorySegments() throws IOException;
+
+	Collection<MemorySegment> requestMemorySegments(int numRequiredBuffers) throws IOException;


If we only consider the semantics of MemorySegmentProvider interface, the requestMemorySegments() is a bit redundant, which could be replaced by requestMemorySegments(int numRequiredBuffers). The only reason we keep both methods is that it seems not elegant to maintain networkBuffersPerChannel in SingleInputGate.

TBH, I do not think it is worth defining this new interface not very clean to accommodate the implementation. Because requestMemorySegments() seems blurry for the usage which relies on the default value, and we should also consider if the default value is 0, whether the returned collection is null or empty size.
In addition, requestMemorySegments(int numRequiredBuffers) is only used for tests ATM.

I think MemorySegmentProvider makes sense from point of interface segregation principle. This interface encapsulates segment management concern which is the only thing that remote channel needs atm. NetworkBufferPool looks like implementing 2 separate concerns: BufferPoolFactory and MemorySegmentProvider.

As for requestMemorySegments(int numRequiredBuffers), you are right, it is more clear to remove it from the interface and adjust the tests with explicit default value, e.g. 1.

We are indeed in the lack of semantics of requesting/releasing batch of buffers before, and I think it is reasonable to extend these methods.

SegmentProvider is not really needed because RemoteInputChannel still needs construct the Buffer internally after requesting MemorySegment.

We could extend the general BufferProvider interface to provide method of requestBuffers(int numRequiredBuffers, BufferRecycler), and RemoteInputChannel already implements BufferRecycler interface.

We could also extend the general BufferRecycler interface to provide method of recycle(Collection<MemorySegment>).

To do so we make use of the existing semantics and architecture. But the only concern is we would affect the previous implementations to bring more changes.

zhijiangW · 2019-05-13T06:10:50Z

...me/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java

@@ -155,8 +155,8 @@
 	 */
 	private BufferPool bufferPool;

-	/** Global network buffer pool to request and recycle exclusive buffers (only for credit-based). */
-	private NetworkBufferPool networkBufferPool;
+	/** Global global memory segment provider to request and recycle exclusive buffers (only for credit-based). */


repeated global

zhijiangW · 2019-05-13T06:18:26Z

...me/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java

@@ -535,7 +510,7 @@ public void requestPartitions() throws IOException, InterruptedException {
 			throws IOException, InterruptedException {
 		while (true) {
 			synchronized (inputChannelsWithData) {
-				while (inputChannelsWithData.size() == 0) {
+				while (inputChannelsWithData.isEmpty()) {


why make this change?

zhijiangW · 2019-05-13T06:21:53Z

...main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateFactory.java

-					metrics
-				);
+					metrics,
+					memorySegmentProvider);


Whether to separate line for last ); which keeps the consistent with local and unknown cases?

zhijiangW · 2019-05-13T06:36:12Z

...c/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java

+	static RemoteInputChannel createRemoteInputChannel(
+		SingleInputGate inputGate, MemorySegmentProvider memorySegmentProvider) throws Exception {
+
+		return createRemoteInputChannel(inputGate, mock(PartitionRequestClient.class), 0, 0, memorySegmentProvider);


It is better to avoid mock PartitionRequestClient here.

zhijiangW · 2019-05-13T06:40:01Z

...c/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java

@@ -237,7 +254,8 @@ static RemoteInputChannel createRemoteInputChannel(
 			SingleInputGate inputGate,
 			PartitionRequestClient client,
 			int initialBackoff,
-			int maxBackoff) throws Exception {
+			int maxBackoff,
+			MemorySegmentProvider memorySegmentProvider) throws Exception {
 		final ConnectionManager connectionManager = mock(ConnectionManager.class);
 		when(connectionManager.createPartitionRequestClient(any(ConnectionID.class)))


There are already too many utils here to construct the RemoteInputChannel, but we still miss the units of {SingleInputGate, PartitionRequestClient, MemorySegmentProvider} which could avoid pass createRemoteInputChannel(inputGate, client, 0, 0, networkBufferPool) in CreditBasedPartitionRequestClientHandlerTest.

Maybe we need a separate hotfix for refactoring these parts to make connectionManager.createPartitionRequestClient call in separate tests if necessary, then we only need one type to construct RemoteInputChannel.

zhijiangW · 2019-05-13T06:41:39Z

...rc/test/java/org/apache/flink/runtime/io/network/partition/consumer/InputChannelBuilder.java

@@ -46,6 +48,7 @@
 	private int maxBackoff = 0;
 	private InputChannelMetrics metrics =
 		new InputChannelMetrics(UnregisteredMetricGroups.createUnregisteredTaskMetricGroup());


Here we could use InputChannelTestUtils.newUnregisteredInputChannelMetrics

zhijiangW · 2019-05-13T06:43:36Z

...test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java

@@ -1010,8 +1032,9 @@ private RemoteInputChannel createRemoteInputChannel(

 		return InputChannelBuilder.newBuilder()
 			.setConnectionManager(connectionManager)
-			.setInitialBackoff(initialBackoff)
+			.setInitialBackoff(initialBackOff)


No need change?

zhijiangW · 2019-05-13T06:46:28Z

...test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java

+	private RemoteInputChannel createRemoteInputChannel(
+		SingleInputGate inputGate,
+		PartitionRequestClient partitionRequestClient,
+		int initialBackOff,


lowercase off to keep consistent with maxBackoff?

zhijiangW · 2019-05-13T06:56:04Z

...test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java

+		throws IOException, InterruptedException {
+
+		return createRemoteInputChannel(
+				inputGate, mock(PartitionRequestClient.class), 0, 0, memorySegmentProvider);


better avoid mock PartitionRequestClient.

The same issue for creating RemoteInputChannel in various units, if we do not need unify the createPartitionRequestClient with the constructing of RemoteInputChannel, all these createRemoteInputChannel could be covered by one InputChannelBuilder.

zhijiangW · 2019-05-13T06:59:50Z

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java

 		checkArgument(bufferPool.getNumberOfRequiredMemorySegments() >= getNumberOfSubpartitions(),
-				"Bug in result partition setup logic: Buffer pool has not enough guaranteed buffers for this result partition.");
+			"Bug in result partition setup logic: Buffer pool has not enough guaranteed buffers for this result partition.");

 		checkState(this.bufferPool == null, "Bug in result partition setup logic: Already registered buffer pool.");


this check should be put before resultPartitionBufferPoolFactory.create

zhijiangW · 2019-05-13T07:11:51Z

flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java

@@ -836,6 +833,21 @@ else if (transitionState(current, ExecutionState.FAILED, t)) {
 		}
 	}

+	private void setupPartionsAndGates() throws IOException {


why not remove this method to use the following static directly?

zhijiangW · 2019-05-13T07:15:34Z

flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java

@@ -262,7 +268,29 @@ public void testExecutionFailsInBlobsMissing() throws Exception {
 	}

 	@Test
-	public void testExecutionFailsInNetworkRegistration() throws Exception {
+	public void testExecutionFailsInNetworkRegistrationFprPartitions() throws Exception {


zhijiangW

Thanks for the work for these refactoring @azagrebin .

For the commits arrangement I have two concerns:

The hotfix of Introduce NetworkEnvironment.create factory should be put firstly because the other ones are relying on it. And the motivation for introducing static creation is not for tests purpose. I think it avoids the complex logics of constructing the factories in the constructor which could initialize the vars directly in easy way.
The hotfix of Introduce ResultPartitionBufferPoolFactory and Introduce InputGateBufferPoolFactory could be squashed with the last two setup commits, because they are only the temporary states, not final ones.

In addition, I have some other concerns for the motivation of introducing related factories:

PartitionBufferPoolFactory: I think it is no need to define this new factory for creating BufferPool, because we already have BufferPoolFactory for doing this, and the NetworkBufferPool implementation is just used for creating BufferPool for partition/gate. We might extend the BufferPoolFactory#createBufferPool to achieve the same motivation and make SingleInputGate/ResultPartition only sees the BufferPoolFactory instead of NetworkBufferPool. The new introduced PartitionBufferPoolFactory would make the relationship more complex and difficult to distinguish the semantic scope. In other words, I do not think two ShuffleService implementations would create the same SingleInputGate/ResultPartition in future. So even though the SingleInputGate sees the NetworkBufferPool which belongs to NetworkEnvironment ATM, it still might make sense, or we could break this tie if necessary in future.
ResultPartitionFactory: It seems reasonable to add the factory layer between ShuffleService and ResultPartitionWriter as one option. The current factory is just wrapping some components in NetworkEnvironment, but these components are still needed to be maintained separately in NetworkEnvironment for other usages. So I am not sure the specific advantages from the architecture view to do so except for the form view.

zhijiangW · 2019-05-13T08:14:54Z

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/PartitionBufferPoolFactory.java

+
+import java.io.IOException;
+
+/** factory of {@link BufferPool} for network partitions or channels. */


This comment annotation /** */ is used for variable, and the class should be like following

/** * */

I am not sure of it.

zhijiangW · 2019-05-17T08:34:40Z

...runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionTest.java

@@ -215,10 +215,9 @@ private void testReleaseMemory(final ResultPartitionType resultPartitionType) th
 		final int numAllBuffers = 10;
 		final NetworkEnvironment network = new NetworkEnvironmentBuilder()
 			.setNumNetworkBuffers(numAllBuffers).build();
-		final ResultPartitionConsumableNotifier notifier = new NoOpResultPartitionConsumableNotifier();
-		final ResultPartition resultPartition = createPartition(notifier, resultPartitionType, false);
+		final ResultPartition resultPartition = createPartition(network, resultPartitionType, 1);


Actually the default value 1 is not needed, just because we could not give all the possible units of parameters in PartitionTestUtils. So it might be better to use ResultPartitionBuilder directly to delete intermediate PartitionTestUtils as I mentioned above.

zhijiangW · 2019-05-17T10:55:32Z

...test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java

+			config.isCreditBased(),
+			config.networkBuffersPerChannel(),
+			config.floatingNetworkBuffersPerGate(),
+			numberOfChannels, partitionType);


separate line for numberOfChannels and partitionType?

zhijiangW · 2019-05-17T11:09:43Z

...time/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionBuilder.java

+ */
+public class ResultPartitionBuilder {
+
+	private JobID jobId = new JobID();


should be final for these vars.

zhijiangW · 2019-05-17T11:24:16Z

Thanks for the updates @azagrebin ! It seems much better now.

I have other two concerns:

It is really nice now to introduce some builders for constructing ResultPartition, SingleInputGate and InputChannel. But there still exits various utils in tests to support various of parameter units for creating these items. It might bring unnecessary changes every time when make changes for partition/gate/channel. Even these scattered utils make hard to maintain and still could not provide everything. So I want to remove all these unnecessary utils in tests to unified use the proposed builder, maybe not in the scope of this PR.
I was ever a bit confused of the proposed MemorySegmentProvider interface, because I think the new introduced methods should be extended in existing BufferProvider and BufferRecycler to support the batch of buffer requesting/recycle. But now I think it seems make sense to do as now, because it is very simple to let NetworkBufferPool implement the new interface and NetworkBufferPool actually has the corresponding functions internally before.

…n of SingleInputGate in tests

…n of ResultPartition in tests

…f InputChannels in tests

azagrebin · 2019-05-18T14:17:37Z

Thanks for the reviews @zentol @zhijiangW , I addressed the comments

zentol · 2019-05-20T06:10:38Z

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java

@@ -115,7 +115,11 @@ public static NetworkEnvironment create(
 			TaskEventPublisher taskEventPublisher,
 			MetricGroup metricGroup,
 			IOManager ioManager) {
-		NettyConfig nettyConfig = checkNotNull(config).nettyConfig();
+		checkNotNull(ioManager);


the check at L 136 is now reedundant

pnowojski

Thanks for the change. I have left couple of smaller comments. Mostly it LGTM.

I do not see a problems with removing this one extra lock, we should be fine without it.

pnowojski · 2019-05-20T09:21:57Z

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java

-		ResultPartitionFactory resultPartitionFactory,
-		SingleInputGateFactory singleInputGateFactory) {
-
+			NetworkEnvironmentConfiguration config,


I think you have squashed/amended/fixed ups wrong commits here. Why is this commit changing formatting?

pnowojski · 2019-05-20T09:23:24Z

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java

@@ -191,7 +198,7 @@ public void registerTask(Task task) throws IOException {
 			}

 			for (final ResultPartition partition : producedPartitions) {
-				setupPartition(partition);
+				partition.setup();


Could you explain more in the commit message of [FLINK-12331][network] Refactor NetworkEnvironment#setupPartition() to ResultPartitionWriter#setup() what is this commit supposed to achieve?

pnowojski · 2019-05-20T09:31:47Z

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java

-	}
-
-	@VisibleForTesting
-	public void setupInputGate(SingleInputGate gate) throws IOException {


Ditto, about commit message: please explain more why & what are you trying to achieve.

pnowojski · 2019-05-20T09:43:18Z

...ntime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateFairnessTest.java

@@ -327,7 +327,8 @@ public FairnessVerifyingInputGate(
 				boolean isCreditBased) {

 			super(owningTaskName, jobId, consumedResultId, ResultPartitionType.PIPELINED,
-				consumedSubpartitionIndex, numberOfInputChannels, taskActions, new SimpleCounter(), isCreditBased);
+				consumedSubpartitionIndex, numberOfInputChannels, taskActions, new SimpleCounter(),
+				isCreditBased, () -> null);


null?

either:

if you are not expecting this supplier function to be ever called, throw new UnsupportedOperationException().

if null is supposed to be supported, change it to Optional.

I would prefer throw new UnsupportedOperationException(). There is no need for Optional in production code.

In that case throw new UnsupportedOperationException is of course better :)

pnowojski · 2019-05-20T09:45:39Z

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java


 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

+import javax.annotation.Nonnull;


nit: I think you could drop Nonnull. We should follow an implicit assumption that everything is Nonnull unless marked differently.

pnowojski · 2019-05-20T09:47:27Z

...time/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionBuilder.java

+
+	private int floatingNetworkBuffersPerGate = 1;
+
+	private FunctionWithException<BufferPoolOwner, BufferPool, IOException> bufferPoolFactory;


Optional instead of Nullable

Is it related to the previous comment about UnsupportedOperationException?

I think not. This builder is later performing != null check on this field

ok, I see, done

azagrebin · 2019-05-20T12:37:02Z

Thanks for the review @pnowojski, I've addressed comments.

zhijiangW

Thanks for the updates. @azagrebin LGTM!

pnowojski

LGTM % address comments commit should be probably squashed with some previous commit(s)

…method

…nel to assign exclusive segments

…RemoteInputChannel

…o ResultPartitionWriter#setup() Move partition setup from NetworkEnvironment to ResultPartition. This eliminates tie between Task and NetworkEnvironment. Task does not need depend on NetworkEnvironment and can trigger setup from ResultPartitionWriter interface.

…o InputGate#setup() Move input gate setup from NetworkEnvironment to InputGate. This eliminates tie between Task and NetworkEnvironment. Task does not need depend on NetworkEnvironment and can trigger setup from InputGate directly.

rmetzger added the review=description? label May 10, 2019

rmetzger added the component=Runtime/Network label May 10, 2019

azagrebin force-pushed the FLINK-12331-az branch 5 times, most recently from 6f664a3 to ca96b3f Compare May 12, 2019 18:57

zhijiangW reviewed May 13, 2019

View reviewed changes

zhijiangW requested changes May 13, 2019

View reviewed changes

zhijiangW reviewed May 13, 2019

View reviewed changes

azagrebin force-pushed the FLINK-12331-az branch from ca96b3f to 98d9524 Compare May 13, 2019 13:21

zhijiangW reviewed May 17, 2019

View reviewed changes

zhijiangW and others added 3 commits May 18, 2019 13:56

[hotfix][tests][network] Introduce SingleInputGateBuilder for creatio…

d4d1b74

…n of SingleInputGate in tests

[hotfix][tests][network] Introduce ResultPartitionBuilder for creatio…

385d3b1

…n of ResultPartition in tests

[hotfix][tests][network] Introduce InputChannelBuilder for creation o…

8ee1523

…f InputChannels in tests

azagrebin force-pushed the FLINK-12331-az branch from a2d51ab to daa897f Compare May 18, 2019 14:16

azagrebin force-pushed the FLINK-12331-az branch from daa897f to 477b0f7 Compare May 18, 2019 16:15

zentol reviewed May 20, 2019

View reviewed changes

azagrebin force-pushed the FLINK-12331-az branch from 477b0f7 to 8b4325f Compare May 20, 2019 08:45

pnowojski requested changes May 20, 2019

View reviewed changes

azagrebin force-pushed the FLINK-12331-az branch from 8b4325f to 0a0edfe Compare May 20, 2019 12:34

azagrebin force-pushed the FLINK-12331-az branch from 0a0edfe to 2c32af1 Compare May 20, 2019 14:19

zhijiangW approved these changes May 20, 2019

View reviewed changes

pnowojski approved these changes May 20, 2019

View reviewed changes

azagrebin added 2 commits May 20, 2019 17:10

[hotfix][tests][network] Introduce NetworkEnvironment.create factory …

42a4bc6

…method

[hotfix][network] Introduce ResultPartitionFactory

6113cc2

azagrebin force-pushed the FLINK-12331-az branch from 2c32af1 to e2299e6 Compare May 21, 2019 08:14

azagrebin mentioned this pull request May 21, 2019

[FLINK-12570] Switch Task from ResultPartition to ResultPartitionWriter interface #8494

Merged

azagrebin added 5 commits May 21, 2019 14:03

[hotfix][network] Introduce SingleInputGateFactory

c782302

[hotfix][network] Introduce MemorySegmentProvider for RemoteInputChan…

509dd50

…nel to assign exclusive segments

[hotfix][network] Move MemorySegmentProvider from SingleInputGate to …

868d28d

…RemoteInputChannel

azagrebin force-pushed the FLINK-12331-az branch from e2299e6 to 4124afe Compare May 21, 2019 12:05

azagrebin mentioned this pull request May 21, 2019

[FLINK-12530][network] Move Task.inputGatesById to NetworkEnvironment #8463

Merged

pnowojski merged commit 3d81425 into apache:master May 22, 2019

		@@ -108,6 +112,7 @@ private NetworkEnvironment(
		this.taskEventPublisher = taskEventPublisher;
		this.isShutdown = false;


		import java.io.IOException;

		/** factory of {@link BufferPool} for network partitions or channels. */


		private int floatingNetworkBuffersPerGate = 1;

		private FunctionWithException<BufferPoolOwner, BufferPool, IOException> bufferPoolFactory;

[FLINK-12331] Introduce partition/gate setup to decouple task registration with NetworkEnvironment #8416

[FLINK-12331] Introduce partition/gate setup to decouple task registration with NetworkEnvironment #8416

Conversation

azagrebin commented May 10, 2019

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented May 10, 2019 • edited

Review Progress

azagrebin commented May 10, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhijiangW left a comment

Choose a reason for hiding this comment

zhijiangW May 13, 2019 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhijiangW commented May 17, 2019

azagrebin commented May 18, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pnowojski left a comment • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pnowojski May 20, 2019 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

azagrebin commented May 20, 2019 • edited

zhijiangW left a comment

Choose a reason for hiding this comment

pnowojski left a comment

Choose a reason for hiding this comment

flinkbot commented May 10, 2019 •

edited

zhijiangW May 13, 2019 •

edited

pnowojski left a comment •

edited

pnowojski May 20, 2019 •

edited

azagrebin commented May 20, 2019 •

edited