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 2022/10/13 19:15:12 UTC
[GitHub] [flink] pnowojski opened a new pull request, #21056: [FLINK-29379][datastream] Example of using Configuration to store ExecutionConfig fields
pnowojski opened a new pull request, #21056:
URL: https://github.com/apache/flink/pull/21056
This is a draft/PoC PR to show a proposal how to use `Configuration` to store `ExecutionConfig` fields.
## 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, Kubernetes/Yarn, 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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a diff in pull request #21056: [FLINK-29379][datastream] Example of using Configuration to store ExecutionConfig fields
Posted by GitBox <gi...@apache.org>.
zentol commented on code in PR #21056:
URL: https://github.com/apache/flink/pull/21056#discussion_r995676246
##########
flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java:
##########
@@ -756,7 +811,27 @@ public GlobalJobParameters getGlobalJobParameters() {
*/
public void setGlobalJobParameters(GlobalJobParameters globalJobParameters) {
Preconditions.checkNotNull(globalJobParameters, "globalJobParameters shouldn't be null");
- this.globalJobParameters = globalJobParameters;
+ configuration.set(
+ GLOBAL_JOB_PARAMETERS_EXECUTION_CONFIG, convertToString(globalJobParameters));
+ }
+
+ private static String convertToString(final Serializable object) {
+ try (final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ ObjectOutputStream oos = new ObjectOutputStream(baos)) {
+ oos.writeObject(object);
+ return Base64.getEncoder().encodeToString(baos.toByteArray());
+ } catch (final IOException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ private static <T extends Serializable> T convertFrom(final String objectAsString) {
+ final byte[] data = Base64.getDecoder().decode(objectAsString);
+ try (final ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(data))) {
+ return (T) ois.readObject();
Review Comment:
This won't work if the GlobalJobParamaters are a user-defined class. You'll need to somehow provide the EC with the usercode CL.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a diff in pull request #21056: [FLINK-29379][datastream] Migrate most of the ExecutionConfig fields to Configuration
Posted by GitBox <gi...@apache.org>.
zentol commented on code in PR #21056:
URL: https://github.com/apache/flink/pull/21056#discussion_r1012905291
##########
flink-core/src/main/java/org/apache/flink/configuration/PipelineOptions.java:
##########
@@ -95,7 +95,7 @@ public class PipelineOptions {
public static final ConfigOption<Duration> AUTO_WATERMARK_INTERVAL =
key("pipeline.auto-watermark-interval")
.durationType()
- .defaultValue(Duration.ZERO)
+ .defaultValue(Duration.ofMillis(200))
Review Comment:
need to regenerate the docs
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a diff in pull request #21056: [FLINK-29379][datastream] Example of using Configuration to store ExecutionConfig fields
Posted by GitBox <gi...@apache.org>.
zentol commented on code in PR #21056:
URL: https://github.com/apache/flink/pull/21056#discussion_r996942242
##########
flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java:
##########
@@ -756,7 +811,27 @@ public GlobalJobParameters getGlobalJobParameters() {
*/
public void setGlobalJobParameters(GlobalJobParameters globalJobParameters) {
Preconditions.checkNotNull(globalJobParameters, "globalJobParameters shouldn't be null");
- this.globalJobParameters = globalJobParameters;
+ configuration.set(
+ GLOBAL_JOB_PARAMETERS_EXECUTION_CONFIG, convertToString(globalJobParameters));
+ }
+
+ private static String convertToString(final Serializable object) {
+ try (final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ ObjectOutputStream oos = new ObjectOutputStream(baos)) {
+ oos.writeObject(object);
+ return Base64.getEncoder().encodeToString(baos.toByteArray());
+ } catch (final IOException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ private static <T extends Serializable> T convertFrom(final String objectAsString) {
+ final byte[] data = Base64.getDecoder().decode(objectAsString);
+ try (final ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(data))) {
+ return (T) ois.readObject();
Review Comment:
sounds good to me. (I also don't see a good alternative :see_no_evil: )
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a diff in pull request #21056: [FLINK-29379][datastream] Migrate most of the ExecutionConfig fields to Configuration
Posted by GitBox <gi...@apache.org>.
zentol commented on code in PR #21056:
URL: https://github.com/apache/flink/pull/21056#discussion_r1012914212
##########
flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java:
##########
@@ -1174,6 +1161,16 @@ public void configure(ReadableConfig configuration, ClassLoader classLoader) {
== JobManagerOptions.SchedulerType.AdaptiveBatch));
}
+ @Internal
+ public <T> T get(ConfigOption<T> option) {
+ return configuration.get(option);
+ }
+
+ @Internal
+ public <T> Optional<T> getOptional(ConfigOption<T> option) {
+ return configuration.getOptional(option);
+ }
Review Comment:
hmm. Not quite sold on these yet. I see the long-term benefits of making a switch to a plain `Configuration` easier, but I'm worried that someone will try to use `getOptional` for an option that is never in the ExecutionConfig.
##########
flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java:
##########
@@ -461,14 +461,20 @@ public RestartStrategies.RestartStrategyConfiguration getRestartStrategy() {
}
}
+ /**
+ * TODO: this shouldn't exist and shouldn't pollute public API. Tests should change this via
Review Comment:
This isn't too difficult to fix though, is it? Can't we modify the config via the TableEnvironment in the affected tests?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a diff in pull request #21056: [FLINK-29379][datastream] Migrate most of the ExecutionConfig fields to Configuration
Posted by GitBox <gi...@apache.org>.
zentol commented on code in PR #21056:
URL: https://github.com/apache/flink/pull/21056#discussion_r1013782333
##########
flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java:
##########
@@ -1174,6 +1161,16 @@ public void configure(ReadableConfig configuration, ClassLoader classLoader) {
== JobManagerOptions.SchedulerType.AdaptiveBatch));
}
+ @Internal
+ public <T> T get(ConfigOption<T> option) {
+ return configuration.get(option);
+ }
+
+ @Internal
+ public <T> Optional<T> getOptional(ConfigOption<T> option) {
+ return configuration.getOptional(option);
+ }
Review Comment:
Are there more usages of these methods (or do we intend to add more)? Maybe we don't need such a generic variant in the first place.
(maybe add a `isTaskCancellationTimeoutSet` method instead?)
Alternatively, maybe we could modify the ExecutionConfig post deserialization on the TM side to contain the "final" execution config.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a diff in pull request #21056: [FLINK-29379][datastream] Example of using Configuration to store ExecutionConfig fields
Posted by GitBox <gi...@apache.org>.
pnowojski commented on code in PR #21056:
URL: https://github.com/apache/flink/pull/21056#discussion_r996725697
##########
flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java:
##########
@@ -756,7 +811,27 @@ public GlobalJobParameters getGlobalJobParameters() {
*/
public void setGlobalJobParameters(GlobalJobParameters globalJobParameters) {
Preconditions.checkNotNull(globalJobParameters, "globalJobParameters shouldn't be null");
- this.globalJobParameters = globalJobParameters;
+ configuration.set(
+ GLOBAL_JOB_PARAMETERS_EXECUTION_CONFIG, convertToString(globalJobParameters));
+ }
+
+ private static String convertToString(final Serializable object) {
+ try (final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ ObjectOutputStream oos = new ObjectOutputStream(baos)) {
+ oos.writeObject(object);
+ return Base64.getEncoder().encodeToString(baos.toByteArray());
+ } catch (final IOException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ private static <T extends Serializable> T convertFrom(final String objectAsString) {
+ final byte[] data = Base64.getDecoder().decode(objectAsString);
+ try (final ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(data))) {
+ return (T) ois.readObject();
Review Comment:
Ok, good point. In that case I'm thinking about keeping those fields as they are, and not migrating them to the `configuration` for the time being. This partial migration would be still a step towards the good direction, and making for example the errors for the disabled programatic changes better. WDYT?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski merged pull request #21056: [FLINK-29379][datastream] Migrate most of the ExecutionConfig fields to Configuration
Posted by GitBox <gi...@apache.org>.
pnowojski merged PR #21056:
URL: https://github.com/apache/flink/pull/21056
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] twalthr commented on a diff in pull request #21056: [FLINK-29379][datastream] Example of using Configuration to store ExecutionConfig fields
Posted by GitBox <gi...@apache.org>.
twalthr commented on code in PR #21056:
URL: https://github.com/apache/flink/pull/21056#discussion_r996752994
##########
flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java:
##########
@@ -746,7 +797,11 @@ public boolean isObjectReuseEnabled() {
}
public GlobalJobParameters getGlobalJobParameters() {
Review Comment:
The alternative to @zentol's comment would otherwise be thread context classloader.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a diff in pull request #21056: [FLINK-29379][datastream] Migrate most of the ExecutionConfig fields to Configuration
Posted by GitBox <gi...@apache.org>.
pnowojski commented on code in PR #21056:
URL: https://github.com/apache/flink/pull/21056#discussion_r1012925751
##########
flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java:
##########
@@ -1174,6 +1161,16 @@ public void configure(ReadableConfig configuration, ClassLoader classLoader) {
== JobManagerOptions.SchedulerType.AdaptiveBatch));
}
+ @Internal
+ public <T> T get(ConfigOption<T> option) {
+ return configuration.get(option);
+ }
+
+ @Internal
+ public <T> Optional<T> getOptional(ConfigOption<T> option) {
+ return configuration.getOptional(option);
+ }
Review Comment:
Yeah, but the only alternative is to replace them with `boolean ExecutionConfig#isPresent(ConfigOption<T>)`. Take a look how I need to use those methods in `Task.java` :(
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a diff in pull request #21056: [FLINK-29379][datastream] Example of using Configuration to store ExecutionConfig fields
Posted by GitBox <gi...@apache.org>.
zentol commented on code in PR #21056:
URL: https://github.com/apache/flink/pull/21056#discussion_r995678770
##########
flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java:
##########
@@ -148,6 +148,19 @@ public class ExecutionConfig implements Serializable, Archiveable<ArchivedExecut
.defaultValue(-1)
.withDescription(
"Should no longer be used because it is subsumed by RestartStrategyConfiguration");
+
+ /**
+ * Conceptually this is the same as {@link PipelineOptions#GLOBAL_JOB_PARAMETERS}, but it's
+ * stored/serialized in the configuration very differently. Use with care... First {@link
+ * PipelineOptions#GLOBAL_JOB_PARAMETERS} is read from the config, converted into {@link
+ * MapBasedJobParameters} and then its stored in the {@link #configuration} via {@link
+ * #GLOBAL_JOB_PARAMETERS_EXECUTION_CONFIG}. User can override this value via {@link
+ * ExecutionConfig#setGlobalJobParameters(GlobalJobParameters)}.
+ */
+ public static final ConfigOption<GlobalJobParameters> GLOBAL_JOB_PARAMETERS_EXECUTION_CONFIG =
+ key(PipelineOptions.GLOBAL_JOB_PARAMETERS.key())
+ .defaultValue(new GlobalJobParameters());
Review Comment:
I guess this is also required for the Kryo serializers and the `restartStrategyConfiguration`.
I nevertheless like the change in general though.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on pull request #21056: [FLINK-29379][datastream] Migrate most of the ExecutionConfig fields to Configuration
Posted by GitBox <gi...@apache.org>.
pnowojski commented on PR #21056:
URL: https://github.com/apache/flink/pull/21056#issuecomment-1301887623
@zentol / @twalthr , could you take a look again? (ignoring FLINK-29807 commits). I've migrated most of the fields, except of those that are problematic.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a diff in pull request #21056: [FLINK-29379][datastream] Migrate most of the ExecutionConfig fields to Configuration
Posted by GitBox <gi...@apache.org>.
pnowojski commented on code in PR #21056:
URL: https://github.com/apache/flink/pull/21056#discussion_r1013854501
##########
flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java:
##########
@@ -1174,6 +1161,16 @@ public void configure(ReadableConfig configuration, ClassLoader classLoader) {
== JobManagerOptions.SchedulerType.AdaptiveBatch));
}
+ @Internal
+ public <T> T get(ConfigOption<T> option) {
+ return configuration.get(option);
+ }
+
+ @Internal
+ public <T> Optional<T> getOptional(ConfigOption<T> option) {
+ return configuration.getOptional(option);
+ }
Review Comment:
As per offline discussion I've replaced those with `toConfiguration()` 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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #21056: [FLINK-29379][datastream] Example of using Configuration to store ExecutionConfig fields
Posted by GitBox <gi...@apache.org>.
flinkbot commented on PR #21056:
URL: https://github.com/apache/flink/pull/21056#issuecomment-1278080912
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1e6eeb6a13210c2ec4add17d5e788d0501d0b5a8",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1e6eeb6a13210c2ec4add17d5e788d0501d0b5a8",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1e6eeb6a13210c2ec4add17d5e788d0501d0b5a8 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a diff in pull request #21056: [FLINK-29379][datastream] Migrate most of the ExecutionConfig fields to Configuration
Posted by GitBox <gi...@apache.org>.
pnowojski commented on code in PR #21056:
URL: https://github.com/apache/flink/pull/21056#discussion_r1014256187
##########
flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java:
##########
@@ -461,14 +461,20 @@ public RestartStrategies.RestartStrategyConfiguration getRestartStrategy() {
}
}
+ /**
+ * TODO: this shouldn't exist and shouldn't pollute public API. Tests should change this via
Review Comment:
Per offline discussion we decided to leave it as it is, due to some quirks and difficulties in fixing in `TableTestUtil`
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on pull request #21056: [FLINK-29379][datastream] Migrate most of the ExecutionConfig fields to Configuration
Posted by GitBox <gi...@apache.org>.
pnowojski commented on PR #21056:
URL: https://github.com/apache/flink/pull/21056#issuecomment-1304725113
@flinkbot run azure
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a diff in pull request #21056: [FLINK-29379][datastream] Example of using Configuration to store ExecutionConfig fields
Posted by GitBox <gi...@apache.org>.
pnowojski commented on code in PR #21056:
URL: https://github.com/apache/flink/pull/21056#discussion_r995070741
##########
flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java:
##########
@@ -148,6 +148,19 @@ public class ExecutionConfig implements Serializable, Archiveable<ArchivedExecut
.defaultValue(-1)
.withDescription(
"Should no longer be used because it is subsumed by RestartStrategyConfiguration");
+
+ /**
+ * Conceptually this is the same as {@link PipelineOptions#GLOBAL_JOB_PARAMETERS}, but it's
+ * stored/serialized in the configuration very differently. Use with care... First {@link
+ * PipelineOptions#GLOBAL_JOB_PARAMETERS} is read from the config, converted into {@link
+ * MapBasedJobParameters} and then its stored in the {@link #configuration} via {@link
+ * #GLOBAL_JOB_PARAMETERS_EXECUTION_CONFIG}. User can override this value via {@link
+ * ExecutionConfig#setGlobalJobParameters(GlobalJobParameters)}.
+ */
+ public static final ConfigOption<GlobalJobParameters> GLOBAL_JOB_PARAMETERS_EXECUTION_CONFIG =
+ key(PipelineOptions.GLOBAL_JOB_PARAMETERS.key())
+ .defaultValue(new GlobalJobParameters());
Review Comment:
Ok. E2e test failure shows that this won't work that easily. `ConfigOption` doesn't support arbitrary types, so I would propose to store it as base64 serialised string.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a diff in pull request #21056: [FLINK-29379][datastream] Migrate most of the ExecutionConfig fields to Configuration
Posted by GitBox <gi...@apache.org>.
pnowojski commented on code in PR #21056:
URL: https://github.com/apache/flink/pull/21056#discussion_r1012923470
##########
flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java:
##########
@@ -461,14 +461,20 @@ public RestartStrategies.RestartStrategyConfiguration getRestartStrategy() {
}
}
+ /**
+ * TODO: this shouldn't exist and shouldn't pollute public API. Tests should change this via
Review Comment:
Ahhh, I've tried to fix this but I couldn't do it easily. One place was easy, but the other three were not :(
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on pull request #21056: [FLINK-29379][datastream] Migrate most of the ExecutionConfig fields to Configuration
Posted by GitBox <gi...@apache.org>.
pnowojski commented on PR #21056:
URL: https://github.com/apache/flink/pull/21056#issuecomment-1304725077
@flinkbot run azur
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a diff in pull request #21056: [FLINK-29379][datastream] Example of using Configuration to store ExecutionConfig fields
Posted by GitBox <gi...@apache.org>.
zentol commented on code in PR #21056:
URL: https://github.com/apache/flink/pull/21056#discussion_r996941898
##########
flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java:
##########
@@ -746,7 +797,11 @@ public boolean isObjectReuseEnabled() {
}
public GlobalJobParameters getGlobalJobParameters() {
Review Comment:
I agree that we should change this to a simple `Map<String, String>` (and force the map to not required the user CL), or maybe a `Configuration`.
IMO this API was never intuitive in the first place.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a diff in pull request #21056: [FLINK-29379][datastream] Example of using Configuration to store ExecutionConfig fields
Posted by GitBox <gi...@apache.org>.
pnowojski commented on code in PR #21056:
URL: https://github.com/apache/flink/pull/21056#discussion_r996830846
##########
flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java:
##########
@@ -756,7 +811,27 @@ public GlobalJobParameters getGlobalJobParameters() {
*/
public void setGlobalJobParameters(GlobalJobParameters globalJobParameters) {
Preconditions.checkNotNull(globalJobParameters, "globalJobParameters shouldn't be null");
- this.globalJobParameters = globalJobParameters;
+ configuration.set(
+ GLOBAL_JOB_PARAMETERS_EXECUTION_CONFIG, convertToString(globalJobParameters));
+ }
+
+ private static String convertToString(final Serializable object) {
+ try (final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ ObjectOutputStream oos = new ObjectOutputStream(baos)) {
+ oos.writeObject(object);
+ return Base64.getEncoder().encodeToString(baos.toByteArray());
+ } catch (final IOException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ private static <T extends Serializable> T convertFrom(final String objectAsString) {
+ final byte[] data = Base64.getDecoder().decode(objectAsString);
+ try (final ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(data))) {
+ return (T) ois.readObject();
Review Comment:
I think in that case I would prefer to leave those problematic fields as they are and migrate only the non java-serialized fields.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] twalthr commented on a diff in pull request #21056: [FLINK-29379][datastream] Example of using Configuration to store ExecutionConfig fields
Posted by GitBox <gi...@apache.org>.
twalthr commented on code in PR #21056:
URL: https://github.com/apache/flink/pull/21056#discussion_r996751165
##########
flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java:
##########
@@ -746,7 +797,11 @@ public boolean isObjectReuseEnabled() {
}
public GlobalJobParameters getGlobalJobParameters() {
Review Comment:
We should solve this differently. `GlobalJobParameters` should be a simple `Map<String, String>`. Maybe we keep those fields in `ExecutionConfig` (not backed by Configuration) and deprecate the corresponding setters and getters and replace them with better long-term API.
Even our examples show how to extend from global job parameters:
https://github.com/apache/flink/blob/master/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/TopSpeedWindowing.java#L83
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org