[FLINK-14070] [runtime] Use TimeUtils to parse duration configs in flink-runtime #9745

zhuzhurk · 2019-09-23T08:25:14Z

What is the purpose of the change

TimeUtils now can parse all duration configs supported by scala FiniteDuration.
And we'd like to use it to replace scala Duration for duration config parsing.
This is one step to make flink-runtime scala free.

Brief change log

The duration configs in flink-runtime are now parsed using TimeUtils#parseDuration

Verifying this change

This change is already covered by existing tests of changed components.

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)

flinkbot · 2019-09-23T08:27:05Z

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 8097abd (Wed Oct 16 08:28:53 UTC 2019)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

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 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

flinkbot · 2019-09-23T08:30:22Z

CI report:

d4ab076 : FAILURE Build
d64b99e : FAILURE Build
cdc9812 : SUCCESS Build
b0cd94b : FAILURE Build
156dcfe : SUCCESS Build
0d518c7 : SUCCESS Build
8097abd : SUCCESS Build

zhuzhurk · 2019-09-23T08:44:10Z

flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala

@@ -49,8 +51,6 @@ import scala.language.postfixOps
 object AkkaUtils {


The parsing of akka configs are not replaced by TimeUtils.parseDuration(..), since they are directly used for ActorSystem creation.
The only exception is "akka.ask.timeout" which is also widely used as Flink RPC timeout config. So I parsed it with TimeUtils only when it is used by Flink components (in AkkaUtils#getTimeout). It is still parsed by scala Duration when used by akka (in AkkaUtils#getRemoteAkkaConfig).
The configs "akka.lookup.timeout" and "akka.client.timeout" are actually not akka configs. They are only used by Flink components so the parsing logics for them are updated.

Would it hurt to replace https://github.com/apache/flink/blob/master/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala#L400 with val akkaAskTimeout = Duration(getTimeout(configuration).toMillis, TimeUnit.MILLISECONDS)? The benefit would be that we don't use two mechanisms for parsing. Consequently, these two mechanism don't need to stay in sync in the future.

I gave it a thought again and I think we can replace them.

The concern was that scala Duration supports infinite values like Inf, +Inf, etc. Replacing the parsing may break jobs specifying infinite values. However, since we never mentioned infinite durations in the config descriptors, I think we can treat them as finite durations and throw errors for infinite values.

Besides that, do you think we should parse the other duration configs (e.g. "akka.tcp.timeout") which were treated as string and directly set to Akka configs? Parsing them can help to do an early check, rather than expecting Akka to fail due to bad configs.

A hotfix is added to convert akka duration configs to akka recognizable duration format.
Given that "these two mechanism don't need to stay in sync in the future" and TimeUtils currently supports non-labeled config, I think the hotfix is necessary.

… parts)

…arts)

tillrohrmann

Thanks for reducing flink-runtime's Scala dependency @zhuzhurk. The change looks good to me. I had a couple of minor comments and a question whether we should not also use AkkaUtils#getTimeout in #getRemoteAkkaConfig.

tillrohrmann · 2019-09-24T13:30:32Z

flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala

-
-    new FiniteDuration(duration.toMillis, TimeUnit.MILLISECONDS)
+  def getTimeout(config: Configuration): time.Duration = {
+    return TimeUtils.parseDuration(config.getString(AkkaOptions.ASK_TIMEOUT))


return is not needed

tillrohrmann · 2019-09-24T13:30:40Z

flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala

-
-    new FiniteDuration(duration.toMillis, TimeUnit.MILLISECONDS)
+  def getLookupTimeout(config: Configuration): time.Duration = {
+    return TimeUtils.parseDuration(config.getString(AkkaOptions.LOOKUP_TIMEOUT))


same here with return

tillrohrmann · 2019-09-24T13:30:45Z

flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala

-
-    new FiniteDuration(duration.toMillis, TimeUnit.MILLISECONDS)
+  def getClientTimeout(config: Configuration): time.Duration = {
+    return TimeUtils.parseDuration(config.getString(AkkaOptions.CLIENT_TIMEOUT))


tillrohrmann · 2019-09-24T13:37:23Z

flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala

@@ -49,8 +51,6 @@ import scala.language.postfixOps
 object AkkaUtils {


Would it hurt to replace https://github.com/apache/flink/blob/master/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala#L400 with val akkaAskTimeout = Duration(getTimeout(configuration).toMillis, TimeUnit.MILLISECONDS)? The benefit would be that we don't use two mechanisms for parsing. Consequently, these two mechanism don't need to stay in sync in the future.

…ated parts)

tillrohrmann

Thanks for the fixups @zhuzhurk. LGTM. Merging now.

rmetzger added the review=description? label Sep 23, 2019

rmetzger added the component=Runtime/Configuration label Sep 23, 2019

zhuzhurk commented Sep 23, 2019

View reviewed changes

zhuzhurk force-pushed the FLINK_14070_duration_config_parsing branch 3 times, most recently from cdc9812 to b0cd94b Compare September 24, 2019 08:47

zhuzhurk added 2 commits September 24, 2019 17:54

[FLINK-14070] Use TimeUtils to parse duration configs (akka unrelated…

50332a9

… parts)

[FLINK-14070] Use TimeUtils to parse duration configs (akka related p…

156dcfe

…arts)

zhuzhurk force-pushed the FLINK_14070_duration_config_parsing branch from b0cd94b to 156dcfe Compare September 24, 2019 09:56

tillrohrmann self-assigned this Sep 24, 2019

tillrohrmann requested changes Sep 24, 2019

View reviewed changes

zhuzhurk added 3 commits September 25, 2019 00:30

Fixup [FLINK-14070] Use TimeUtils to parse duration configs (akka rel…

0d518c7

…ated parts)

Fixup [FLINK-14070] Use TimeUtils to parse duration configs (akka rel…

8aa11c7

…ated parts)

[hotfix] Convert Akka duration configs to valid Akka duration format

8097abd

tillrohrmann approved these changes Sep 25, 2019

View reviewed changes

tillrohrmann closed this in 949776b Sep 25, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-14070] [runtime] Use TimeUtils to parse duration configs in flink-runtime #9745

[FLINK-14070] [runtime] Use TimeUtils to parse duration configs in flink-runtime #9745

zhuzhurk commented Sep 23, 2019

flinkbot commented Sep 23, 2019 •

edited

flinkbot commented Sep 23, 2019 •

edited

zhuzhurk Sep 23, 2019 •

edited

tillrohrmann Sep 24, 2019

zhuzhurk Sep 24, 2019

zhuzhurk Sep 24, 2019

tillrohrmann left a comment

tillrohrmann Sep 24, 2019

zhuzhurk Sep 24, 2019

tillrohrmann Sep 24, 2019

zhuzhurk Sep 24, 2019

tillrohrmann Sep 24, 2019

zhuzhurk Sep 24, 2019

tillrohrmann Sep 24, 2019

tillrohrmann left a comment

		@@ -49,8 +51,6 @@ import scala.language.postfixOps
		object AkkaUtils {

[FLINK-14070] [runtime] Use TimeUtils to parse duration configs in flink-runtime #9745

[FLINK-14070] [runtime] Use TimeUtils to parse duration configs in flink-runtime #9745

Conversation

zhuzhurk commented Sep 23, 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 Sep 23, 2019 • edited

Automated Checks

Review Progress

flinkbot commented Sep 23, 2019 • edited

CI report:

zhuzhurk Sep 23, 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

tillrohrmann left a 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

tillrohrmann left a comment

Choose a reason for hiding this comment

flinkbot commented Sep 23, 2019 •

edited

flinkbot commented Sep 23, 2019 •

edited

zhuzhurk Sep 23, 2019 •

edited