[FLINK-12706] Introduce ShuffleServiceFactory interface and its configuration #8680

azagrebin · 2019-06-10T19:58:37Z

What is the purpose of the change

Introduce ShuffleService interface with two factory methods for ShuffleMaster and ShuffleEnvironment and its configuration. Implement and configure by default NettyShuffleService.

Brief change log

Introduce ShuffleService interface with two factory methods for ShuffleMaster and ShuffleEnvironment
Introduce configuration core option with the full name of the implementation class and a ShuffleServiceLoader utility class to load the configured ShuffleService implementation from the class path
Implement and configure by default NettyShuffleService.

Verifying this change

Refactoring, 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): (yes)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (np)
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? (docs / JavaDocs)

azagrebin · 2019-06-10T19:59:14Z

The PR is based on #8608 at the moment, only last two commits are to review.

flinkbot · 2019-06-10T20:01:01Z

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 @tillrohrmann [PMC], @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.

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

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-06-10T20:02:22Z

@flinkbot attention @tillrohrmann @zhijiangW

zhijiangW · 2019-06-11T07:56:37Z

flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java

+		.key("shuffle.service.class.name")
+		.defaultValue("org.apache.flink.runtime.io.network.NettyShuffleService")
+		.withDescription("The full name of shuffle service implementation class. " +
+			"The default implentation uses netty network communication, local memory and file system on task executor");


implentation -> implementation

zhijiangW · 2019-06-11T07:58:54Z

flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java

+	/**
+	 * The shuffle service implementation class full name.
+	 */
+	public static final ConfigOption<String> SHUFFLE_SERVICE = ConfigOptions


why not put this option into runtime module? Are there any considerations

On the one hand, it would not hurt to put it into the flink-runtime module because it should be a cluster only option and thus not needed in an API module. On the other hand we also put the NettyShuffleEnvironmentOptions and HistoryServerOptions into flink-core. Thus, one could argue that the current placement is in line with the other options.

zhijiangW · 2019-06-11T08:07:27Z

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

+			TaskEventPublisher taskEventPublisher,
+			MetricGroup metricGroup,
+			IOManager ioManager) {
+		checkNotNull(ioManager);


better to obey the parameters sequence during checkNotNull.
also check for the left taskExecutorLocation and metricGroup?

zhijiangW · 2019-06-11T08:08:54Z

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

+	@VisibleForTesting
+	static NettyShuffleEnvironment createNettyShuffleEnvironment(
+			NettyShuffleEnvironmentConfiguration config,
+			ResourceID taskExecutorLocation,


taskExecutorLocation -> resourceID, because we have the TaskManagerLocation class which might be confused

Could also be called taskExecutorResourceId

zhijiangW · 2019-06-11T08:10:53Z

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

+	}
+
+	private static void registerNetworkMetrics(MetricGroup metricGroup, NetworkBufferPool networkBufferPool) {
+		checkNotNull(metricGroup);


might remove this if this check could be done before in createNettyShuffleEnvironment

zhijiangW · 2019-06-11T08:24:30Z

flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/ShuffleEnvironment.java

+	/**
+	 * Task executor local context of shuffle service used to create {@link ShuffleEnvironment}.
+	 */
+	class ShuffleLocalContext {


This class seems not be used atm. What is the motivation for bringing this?

true, some rebase leftover, it is now ShuffleEnvironmentContext

zhijiangW · 2019-06-11T08:28:56Z

flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/ShuffleService.java

+	 * Factory method to create a specific {@link ShuffleMaster} implementation in job master.
+	 *
+	 * @param configuration Flink configuration
+	 * @return shuffle manager implementation


shuffle manager -> shuffle master

zhijiangW · 2019-06-11T08:35:10Z

flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/ShuffleServiceLoader.java

+/**
+ * Utility to load the pluggable {@link ShuffleService} implementations.
+ */
+public class ShuffleServiceLoader {


It might need a unit test for covering this class

zhijiangW

Thanks for opening this PR @azagrebin !

It looks overall good to me and I left some nit inline comments.

tillrohrmann

Thanks for opening this PR @azagrebin. I had some comments mainly concerning the configuration options which we should resolve before moving forward.

tillrohrmann · 2019-06-13T10:00:08Z

flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java

+	// ------------------------------------------------------------------------
+
+	/**
+	 * The shuffle service implementation class full name.


Class name of the shuffle service implementation to be used by the cluster.

tillrohrmann · 2019-06-13T10:01:33Z

flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java

+	/**
+	 * The shuffle service implementation class full name.
+	 */
+	public static final ConfigOption<String> SHUFFLE_SERVICE = ConfigOptions


Maybe better to call this option SHUFFLE_SERVICE_CLASS

tillrohrmann · 2019-06-13T10:01:54Z

flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java

+	 * The shuffle service implementation class full name.
+	 */
+	public static final ConfigOption<String> SHUFFLE_SERVICE = ConfigOptions
+		.key("shuffle.service.class.name")


I would suggest to use the following key: shuffle-service.class

tillrohrmann · 2019-06-13T10:03:04Z

flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java

+	public static final ConfigOption<String> SHUFFLE_SERVICE = ConfigOptions
+		.key("shuffle.service.class.name")
+		.defaultValue("org.apache.flink.runtime.io.network.NettyShuffleService")
+		.withDescription("The full name of shuffle service implementation class. " +


The full class name of the shuffle service implementation to be used by the cluster.

tillrohrmann · 2019-06-13T10:05:03Z

flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java

+		.key("shuffle.service.class.name")
+		.defaultValue("org.apache.flink.runtime.io.network.NettyShuffleService")
+		.withDescription("The full name of shuffle service implementation class. " +
+			"The default implentation uses netty network communication, local memory and file system on task executor");


The default implementation uses Netty for network communication and local memory as well disk space to store results on a TaskExecutor.

tillrohrmann · 2019-06-13T11:37:44Z

flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/ShuffleService.java

+	ShuffleMaster<SD> createShuffleMaster(Configuration configuration);
+
+	/**
+	 * Factory method to create a specific local {@link ShuffleEnvironment} implementation in task executor.


remove in task executor

tillrohrmann · 2019-06-13T11:40:23Z

flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/ShuffleServiceLoader.java

+		return shuffleService.orElseThrow(() -> new FlinkException(
+			String.format(
+				"Could not instantiate the ShuffleService '%s'. " +
+					"Please make sure that this class is on your class path.",


Indentation seems wrong.

This is still the same argument, the string is broken by concatenation, I suggest we leave it this way to distinguish it from the next argument.

tillrohrmann · 2019-06-13T11:42:06Z

flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/ShuffleServiceLoader.java

+/**
+ * Utility to load the pluggable {@link ShuffleService} implementations.
+ */
+public class ShuffleServiceLoader {


Could simply be an enum since it only contains static methods.

tillrohrmann · 2019-06-13T11:44:02Z

flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/ShuffleServiceLoader.java

+				ShuffleService.class,
+				classLoader));
+		} catch (ClassNotFoundException e) {
+			return Optional.empty();


Why are we swallowing this exception here? It looks as if we recreate it at the call site of lookupEmbeddedShuffleServices if the result is empty. I think it would be better to simply return ShuffleService<?, ?, ?> and to allow this method to throw a ClassNotFoundException.

tillrohrmann · 2019-06-13T11:55:30Z

flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/ShuffleServiceLoader.java

+				shuffleServiceClassName)));
+	}
+
+	private static Optional<ShuffleService<?, ?, ?>> lookupEmbeddedShuffleServices(String shuffleServiceClassName) {


This method has some code duplication with HighAvailabilityServiceUtils#createCustomHAServices. Maybe we could factor a common method out which calls InstantiationUtil.instantiate and creates a FlinkException in case of a ClassNotFoundException.

I will move wrapping of ClassNotFoundException by FlinkException into InstantiationUtil.instantiate.

azagrebin · 2019-06-17T09:01:34Z

Thanks for the review @zhijiangW @tillrohrmann, I have addressed comments

zhijiangW · 2019-06-18T06:47:49Z

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

 */
 public class SingleInputGateFactory {
+	@SuppressWarnings("LoggerInitializedWithForeignClass")
 	private static final Logger LOG = LoggerFactory.getLogger(SingleInputGate.class);


why not use SingleInputGateFactory.class here?

My idea was to preserve the structure of the previous logging messages because the factory was a result of refactoring. I can change this if it is not important.

got it, thanks for the explanation!

zhijiangW · 2019-06-18T06:50:07Z

docs/ops/config.md


 {% include generated/resource_manager_configuration.html %}

+### Shuffle service


Upper case for Service for consistent with others.

zhijiangW · 2019-06-18T06:50:22Z

docs/ops/config.zh.md


 {% include generated/resource_manager_configuration.html %}

+### Shuffle service


ditto: Service

zhijiangW · 2019-06-18T07:36:48Z

flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/ShuffleServiceLoaderTest.java

+import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
+import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
+import org.apache.flink.util.FlinkException;
+import org.junit.Test;


should add an empty line before this import.

zhijiangW · 2019-06-18T07:38:26Z

flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/ShuffleServiceLoaderTest.java

+		configuration.setString(SHUFFLE_SERVICE_FACTORY_CLASS, "org.apache.flink.runtime.shuffle.ShuffleServiceLoaderTest$CustomShuffleServiceFactory");
+		ShuffleServiceFactory<?, ?, ?> shuffleServiceFactory = ShuffleServiceLoader.loadShuffleServiceFactory(configuration);
+		assertThat(
+			"Loaded shuffle service factory is not the default netty implementation",


should adjust the message

zhijiangW · 2019-06-18T07:43:47Z

@azagrebin thanks for the updates and I only left several minor comments.

BTW, the commit of [hotfix][network] Rename taskExecutorLocation should be tagged as fixup instead of hotfix. fixup would be squashed when merging, but hotfix would be merged in a separate commit.

azagrebin · 2019-06-18T08:19:43Z

@zhijiangW
thanks for the review, I have addressed comments.
I actually planned [hotfix][network] Rename taskExecutorLocation as a separate commit because it does somewhat unrelated cleanup.

zhijiangW · 2019-06-18T08:52:37Z

Sorry I thought the renaming issue was caused by this PR in previous commit before.
Yes, if this renaming already existed before this PR, it should be a hotfix here. :)
Thanks for updating, LGTM!

…Id and small fixes

… implementing class is not found in InstantiationUtil.instantiate

…ctory method and its default Netty implementation

…and its default Netty implementation

tillrohrmann

Really good work @azagrebin. LGTM. Addressing my last two comments while merging this PR.

tillrohrmann · 2019-06-20T17:09:17Z

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

 			LOG.info("Shutting down the network environment and its components.");

 			// terminate all network connections
+			//noinspection OverlyBroadCatchBlock


Let's revert this.

tillrohrmann · 2019-06-20T17:12:00Z

flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/ShuffleServiceLoaderTest.java

+/**
+ * Test suite for {@link ShuffleServiceLoader} utility.
+ */
+public class ShuffleServiceLoaderTest {


extends TestLogger is missing

…and its default Netty implementation This closes apache#8680.

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

rmetzger added component=Runtime/Network component=Runtime/Coordination component=Runtime/Configuration labels Jun 10, 2019

zhijiangW reviewed Jun 11, 2019

View reviewed changes

zentol mentioned this pull request Jun 11, 2019

[FLINK-12612][coordination] Track stored partition on the TaskExecutor #8687

Merged

zhijiangW reviewed Jun 11, 2019

View reviewed changes

tillrohrmann self-assigned this Jun 13, 2019

tillrohrmann requested changes Jun 13, 2019

View reviewed changes

azagrebin force-pushed the FLINK-12706 branch 2 times, most recently from 9281b24 to 5e0465f Compare June 17, 2019 09:00

azagrebin mentioned this pull request Jun 17, 2019

[FLINK-12555] Introduce an encapsulated metric group layout for shuffle API #8485

Merged

zhijiangW reviewed Jun 18, 2019

View reviewed changes

azagrebin force-pushed the FLINK-12706 branch from 5e0465f to ac438d2 Compare June 18, 2019 08:11

azagrebin force-pushed the FLINK-12706 branch from ac438d2 to 3976099 Compare June 18, 2019 08:18

azagrebin added 4 commits June 20, 2019 10:34

[hotfix][network] Rename taskExecutorLocation to taskExecutorResource…

ef44324

…Id and small fixes

[hotfix] Convert ClassNotFoundException to FlinkException if the type…

a9c98a1

… implementing class is not found in InstantiationUtil.instantiate

[FLINK-12706] Introduce ShuffleService with the ShuffleEnvironment fa…

642f668

…ctory method and its default Netty implementation

[FLINK-12706] Add the ShuffleMaster factory method to ShuffleService …

7ef50b6

…and its default Netty implementation

azagrebin force-pushed the FLINK-12706 branch from 3976099 to 7ef50b6 Compare June 20, 2019 08:37

tillrohrmann approved these changes Jun 20, 2019

View reviewed changes

tillrohrmann closed this in 318ce57 Jun 20, 2019

sjwiesman pushed a commit to sjwiesman/flink that referenced this pull request Jun 26, 2019

[FLINK-12706] Add the ShuffleMaster factory method to ShuffleService …

9306115

…and its default Netty implementation This closes apache#8680.

azagrebin changed the title ~~[FLINK-12706] Introduce ShuffleService interface and its configuration~~ [FLINK-12706] Introduce ShuffleServiceFactory interface and its configuration Aug 7, 2019


		{% include generated/resource_manager_configuration.html %}

		### Shuffle service

[FLINK-12706] Introduce ShuffleServiceFactory interface and its configuration #8680

[FLINK-12706] Introduce ShuffleServiceFactory interface and its configuration #8680

Uh oh!

Conversation

azagrebin commented Jun 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

Uh oh!

azagrebin commented Jun 10, 2019

Uh oh!

flinkbot commented Jun 10, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Review Progress

Uh oh!

azagrebin commented Jun 10, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zhijiangW left a comment

Choose a reason for hiding this comment

Uh oh!

tillrohrmann left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

azagrebin commented Jun 17, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

flinkbot commented Jun 10, 2019 •

edited

Loading