You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by GitBox <gi...@apache.org> on 2020/01/17 23:55:33 UTC
[GitHub] [samza] kw2542 opened a new pull request #1255: SAMZA-2429: Update
LocalApplicatoinRunner to load full job config from config loader when
present.
kw2542 opened a new pull request #1255: SAMZA-2429: Update LocalApplicatoinRunner to load full job config from config loader when present.
URL: https://github.com/apache/samza/pull/1255
Design:
https://cwiki.apache.org/confluence/display/SAMZA/SEP-23%3A+Simplify+Job+Runner
Changes:
1. Update unit test only public constructor of LocalApplicationRunner to take SamzaApplication and Config as well to be consistent with other constructors.
2. Update private constructor of LocalApplicationRunner to take SamzaApplication and Config.
3. Update private constructor of LocalApplicationRunner to load full job config when config loader is present.
API Changes:
N/A. This is part of a series PRs, detailed information will be provided in the last/main PR.
Upgrade Instructions:
This is part of a series PRs, detailed information will be provided in the last/main PR.
Usage Instructions:
This is part of a series PRs, detailed information will be provided in the last/main PR.
Tests
Unit 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] cameronlee314 commented on a change in pull request #1255:
SAMZA-2429: Update LocalApplicatoinRunner to load full job config from
config loader when present.
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #1255: SAMZA-2429: Update LocalApplicatoinRunner to load full job config from config loader when present.
URL: https://github.com/apache/samza/pull/1255#discussion_r372033238
##########
File path: samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
##########
@@ -150,6 +163,10 @@ static MetadataStoreFactory getDefaultCoordinatorStreamStoreFactory(JobConfig jo
}
private static Optional<CoordinationUtils> getCoordinationUtils(Config config) {
+ if (new JobConfig(config).getConfigLoaderFactory().isPresent()) {
+ config = ConfigUtil.loadConfig(config);
+ }
Review comment:
Same as above about multiple calls to `loadConfig`.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] kw2542 commented on a change in pull request #1255:
SAMZA-2429: Update LocalApplicatoinRunner to load full job config from
config loader when present.
Posted by GitBox <gi...@apache.org>.
kw2542 commented on a change in pull request #1255: SAMZA-2429: Update LocalApplicatoinRunner to load full job config from config loader when present.
URL: https://github.com/apache/samza/pull/1255#discussion_r371990759
##########
File path: samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
##########
@@ -111,27 +112,35 @@ public LocalApplicationRunner(SamzaApplication app, Config config) {
* @param metadataStoreFactory the instance of {@link MetadataStoreFactory} to read and write to coordinator stream.
*/
public LocalApplicationRunner(SamzaApplication app, Config config, MetadataStoreFactory metadataStoreFactory) {
- this(ApplicationDescriptorUtil.getAppDescriptor(app, config), getCoordinationUtils(config), metadataStoreFactory);
+ this(app, config, getCoordinationUtils(config), metadataStoreFactory);
}
/**
* Constructor only used in unit test to allow injection of {@link LocalJobPlanner}
*/
@VisibleForTesting
- LocalApplicationRunner(ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc, Optional<CoordinationUtils> coordinationUtils) {
- this(appDesc, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(appDesc.getConfig())));
+ LocalApplicationRunner(SamzaApplication app, Config config, Optional<CoordinationUtils> coordinationUtils) {
+ this(app, config, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(config)));
}
private LocalApplicationRunner(
Review comment:
Got you, in addition to note this change in PR's description, anywhere else we should notify this change?
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] cameronlee314 commented on a change in pull request #1255:
SAMZA-2429: Update LocalApplicatoinRunner to load full job config from
config loader when present.
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #1255: SAMZA-2429: Update LocalApplicatoinRunner to load full job config from config loader when present.
URL: https://github.com/apache/samza/pull/1255#discussion_r371584072
##########
File path: samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
##########
@@ -111,27 +112,35 @@ public LocalApplicationRunner(SamzaApplication app, Config config) {
* @param metadataStoreFactory the instance of {@link MetadataStoreFactory} to read and write to coordinator stream.
*/
public LocalApplicationRunner(SamzaApplication app, Config config, MetadataStoreFactory metadataStoreFactory) {
- this(ApplicationDescriptorUtil.getAppDescriptor(app, config), getCoordinationUtils(config), metadataStoreFactory);
+ this(app, config, getCoordinationUtils(config), metadataStoreFactory);
}
/**
* Constructor only used in unit test to allow injection of {@link LocalJobPlanner}
*/
@VisibleForTesting
- LocalApplicationRunner(ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc, Optional<CoordinationUtils> coordinationUtils) {
- this(appDesc, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(appDesc.getConfig())));
+ LocalApplicationRunner(SamzaApplication app, Config config, Optional<CoordinationUtils> coordinationUtils) {
+ this(app, config, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(config)));
}
private LocalApplicationRunner(
Review comment:
`LocalApplicationRunner` is kind of in that middle layer: it's not in samza-api and not quite a direct API, but it also can get used by users running standalone and it can be specified in configs as the "app.runner.class".
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] cameronlee314 commented on a change in pull request #1255:
SAMZA-2429: Update LocalApplicatoinRunner to load full job config from
config loader when present.
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #1255: SAMZA-2429: Update LocalApplicatoinRunner to load full job config from config loader when present.
URL: https://github.com/apache/samza/pull/1255#discussion_r371562464
##########
File path: samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
##########
@@ -111,27 +112,35 @@ public LocalApplicationRunner(SamzaApplication app, Config config) {
* @param metadataStoreFactory the instance of {@link MetadataStoreFactory} to read and write to coordinator stream.
*/
public LocalApplicationRunner(SamzaApplication app, Config config, MetadataStoreFactory metadataStoreFactory) {
- this(ApplicationDescriptorUtil.getAppDescriptor(app, config), getCoordinationUtils(config), metadataStoreFactory);
+ this(app, config, getCoordinationUtils(config), metadataStoreFactory);
}
/**
* Constructor only used in unit test to allow injection of {@link LocalJobPlanner}
*/
@VisibleForTesting
- LocalApplicationRunner(ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc, Optional<CoordinationUtils> coordinationUtils) {
- this(appDesc, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(appDesc.getConfig())));
+ LocalApplicationRunner(SamzaApplication app, Config config, Optional<CoordinationUtils> coordinationUtils) {
+ this(app, config, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(config)));
Review comment:
I guess this constructor is just for testing, but it does have a similar change as above where `config` is now just the initial config, but before it was the full 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] cameronlee314 commented on a change in pull request #1255:
SAMZA-2429: Update LocalApplicatoinRunner to load full job config from
config loader when present.
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #1255: SAMZA-2429: Update LocalApplicatoinRunner to load full job config from config loader when present.
URL: https://github.com/apache/samza/pull/1255#discussion_r372033107
##########
File path: samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
##########
@@ -111,29 +112,41 @@ public LocalApplicationRunner(SamzaApplication app, Config config) {
* @param metadataStoreFactory the instance of {@link MetadataStoreFactory} to read and write to coordinator stream.
*/
public LocalApplicationRunner(SamzaApplication app, Config config, MetadataStoreFactory metadataStoreFactory) {
- this(ApplicationDescriptorUtil.getAppDescriptor(app, config), getCoordinationUtils(config), metadataStoreFactory);
+ this(app, config, getCoordinationUtils(config), metadataStoreFactory);
}
/**
* Constructor only used in unit test to allow injection of {@link LocalJobPlanner}
*/
@VisibleForTesting
- LocalApplicationRunner(ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc, Optional<CoordinationUtils> coordinationUtils) {
- this(appDesc, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(appDesc.getConfig())));
+ LocalApplicationRunner(SamzaApplication app, Config config, Optional<CoordinationUtils> coordinationUtils) {
+ this(app, config, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(config)));
}
private LocalApplicationRunner(
- ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc,
+ SamzaApplication app,
+ Config config,
Optional<CoordinationUtils> coordinationUtils,
MetadataStoreFactory metadataStoreFactory) {
- this.appDesc = appDesc;
+ this.appDesc = getApplicationDescriptor(app, config);
this.isAppModeBatch = isAppModeBatch(appDesc.getConfig());
this.coordinationUtils = coordinationUtils;
this.metadataStoreFactory = Optional.ofNullable(metadataStoreFactory);
}
+ @VisibleForTesting
+ static ApplicationDescriptorImpl<? extends ApplicationDescriptor> getApplicationDescriptor(SamzaApplication app, Config config) {
+ return new JobConfig(config).getConfigLoaderFactory().isPresent()
+ ? ApplicationDescriptorUtil.getAppDescriptor(app, ConfigUtil.loadConfig(config))
+ : ApplicationDescriptorUtil.getAppDescriptor(app, config);
+ }
+
@VisibleForTesting
static MetadataStoreFactory getDefaultCoordinatorStreamStoreFactory(JobConfig jobConfig) {
+ if (jobConfig.getConfigLoaderFactory().isPresent()) {
+ jobConfig = new JobConfig(ConfigUtil.loadConfig(jobConfig));
+ }
Review comment:
Should we avoid multiple calls to `ConfigUtil.loadConfig`? You already call it in `getApplicationDescriptor`.
If the config loader logic is complex/heavy (e.g. makes remote calls), then that is going to result in inefficient calls.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] cameronlee314 commented on a change in pull request #1255:
SAMZA-2429: Update LocalApplicatoinRunner to load full job config from
config loader when present.
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #1255: SAMZA-2429: Update LocalApplicatoinRunner to load full job config from config loader when present.
URL: https://github.com/apache/samza/pull/1255#discussion_r372129266
##########
File path: samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
##########
@@ -111,29 +112,41 @@ public LocalApplicationRunner(SamzaApplication app, Config config) {
* @param metadataStoreFactory the instance of {@link MetadataStoreFactory} to read and write to coordinator stream.
*/
public LocalApplicationRunner(SamzaApplication app, Config config, MetadataStoreFactory metadataStoreFactory) {
- this(ApplicationDescriptorUtil.getAppDescriptor(app, config), getCoordinationUtils(config), metadataStoreFactory);
+ this(app, config, getCoordinationUtils(config), metadataStoreFactory);
}
/**
* Constructor only used in unit test to allow injection of {@link LocalJobPlanner}
*/
@VisibleForTesting
- LocalApplicationRunner(ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc, Optional<CoordinationUtils> coordinationUtils) {
- this(appDesc, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(appDesc.getConfig())));
+ LocalApplicationRunner(SamzaApplication app, Config config, Optional<CoordinationUtils> coordinationUtils) {
+ this(app, config, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(config)));
}
private LocalApplicationRunner(
- ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc,
+ SamzaApplication app,
+ Config config,
Optional<CoordinationUtils> coordinationUtils,
MetadataStoreFactory metadataStoreFactory) {
- this.appDesc = appDesc;
+ this.appDesc = getApplicationDescriptor(app, config);
this.isAppModeBatch = isAppModeBatch(appDesc.getConfig());
this.coordinationUtils = coordinationUtils;
this.metadataStoreFactory = Optional.ofNullable(metadataStoreFactory);
}
+ @VisibleForTesting
+ static ApplicationDescriptorImpl<? extends ApplicationDescriptor> getApplicationDescriptor(SamzaApplication app, Config config) {
+ return new JobConfig(config).getConfigLoaderFactory().isPresent()
+ ? ApplicationDescriptorUtil.getAppDescriptor(app, ConfigUtil.loadConfig(config))
+ : ApplicationDescriptorUtil.getAppDescriptor(app, config);
+ }
+
@VisibleForTesting
static MetadataStoreFactory getDefaultCoordinatorStreamStoreFactory(JobConfig jobConfig) {
+ if (jobConfig.getConfigLoaderFactory().isPresent()) {
+ jobConfig = new JobConfig(ConfigUtil.loadConfig(jobConfig));
+ }
Review comment:
I do think it is reasonable trade-off.
One of the challenges is that the `LocalApplicationRunner(SamzaApplication app, Config config)` constructor is overloaded semantically. In one case, `config` is already the full config, and in another case `config` is a partial config which needs to be loaded. And then it may seem like we can't use a single common constructor due to the need to use the full config in the constructor arguments. I do still think there is a way to share a common constructor: instead of passing individual arguments, create a `LocalApplicationRunnerContext` which contains all of the arguments, and then you can build that other object based on what is given to you.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] kw2542 commented on a change in pull request #1255:
SAMZA-2429: Update LocalApplicatoinRunner to load full job config from
config loader when present.
Posted by GitBox <gi...@apache.org>.
kw2542 commented on a change in pull request #1255: SAMZA-2429: Update LocalApplicatoinRunner to load full job config from config loader when present.
URL: https://github.com/apache/samza/pull/1255#discussion_r372125290
##########
File path: samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
##########
@@ -111,27 +112,35 @@ public LocalApplicationRunner(SamzaApplication app, Config config) {
* @param metadataStoreFactory the instance of {@link MetadataStoreFactory} to read and write to coordinator stream.
*/
public LocalApplicationRunner(SamzaApplication app, Config config, MetadataStoreFactory metadataStoreFactory) {
- this(ApplicationDescriptorUtil.getAppDescriptor(app, config), getCoordinationUtils(config), metadataStoreFactory);
+ this(app, config, getCoordinationUtils(config), metadataStoreFactory);
}
/**
* Constructor only used in unit test to allow injection of {@link LocalJobPlanner}
*/
@VisibleForTesting
- LocalApplicationRunner(ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc, Optional<CoordinationUtils> coordinationUtils) {
- this(appDesc, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(appDesc.getConfig())));
+ LocalApplicationRunner(SamzaApplication app, Config config, Optional<CoordinationUtils> coordinationUtils) {
+ this(app, config, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(config)));
}
private LocalApplicationRunner(
Review comment:
I will make sure this to be added in the OSS release when it is out and I will go through our user guide to see if we have any thing about this flow.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] cameronlee314 commented on a change in pull request #1255:
SAMZA-2429: Update LocalApplicatoinRunner to load full job config from
config loader when present.
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #1255: SAMZA-2429: Update LocalApplicatoinRunner to load full job config from config loader when present.
URL: https://github.com/apache/samza/pull/1255#discussion_r371562304
##########
File path: samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
##########
@@ -111,27 +112,35 @@ public LocalApplicationRunner(SamzaApplication app, Config config) {
* @param metadataStoreFactory the instance of {@link MetadataStoreFactory} to read and write to coordinator stream.
*/
public LocalApplicationRunner(SamzaApplication app, Config config, MetadataStoreFactory metadataStoreFactory) {
- this(ApplicationDescriptorUtil.getAppDescriptor(app, config), getCoordinationUtils(config), metadataStoreFactory);
+ this(app, config, getCoordinationUtils(config), metadataStoreFactory);
Review comment:
In the old flow, `getCoordinationUtils` seemed to be called with the full config. In the new flow, `getCoordinationUtils` will get called with the initial config, not the full config. Do you need to load the config before calling `getCoordinationUtils`?
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] cameronlee314 commented on a change in pull request #1255:
SAMZA-2429: Update LocalApplicatoinRunner to load full job config from
config loader when present.
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #1255: SAMZA-2429: Update LocalApplicatoinRunner to load full job config from config loader when present.
URL: https://github.com/apache/samza/pull/1255#discussion_r372031567
##########
File path: samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
##########
@@ -111,27 +112,35 @@ public LocalApplicationRunner(SamzaApplication app, Config config) {
* @param metadataStoreFactory the instance of {@link MetadataStoreFactory} to read and write to coordinator stream.
*/
public LocalApplicationRunner(SamzaApplication app, Config config, MetadataStoreFactory metadataStoreFactory) {
- this(ApplicationDescriptorUtil.getAppDescriptor(app, config), getCoordinationUtils(config), metadataStoreFactory);
+ this(app, config, getCoordinationUtils(config), metadataStoreFactory);
}
/**
* Constructor only used in unit test to allow injection of {@link LocalJobPlanner}
*/
@VisibleForTesting
- LocalApplicationRunner(ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc, Optional<CoordinationUtils> coordinationUtils) {
- this(appDesc, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(appDesc.getConfig())));
+ LocalApplicationRunner(SamzaApplication app, Config config, Optional<CoordinationUtils> coordinationUtils) {
+ this(app, config, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(config)));
}
private LocalApplicationRunner(
Review comment:
Someone will need to remember to add this as a release note, and if we have information in the user guide about this flow, then that will need to be updated.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] kw2542 commented on a change in pull request #1255:
SAMZA-2429: Update LocalApplicatoinRunner to load full job config from
config loader when present.
Posted by GitBox <gi...@apache.org>.
kw2542 commented on a change in pull request #1255: SAMZA-2429: Update LocalApplicatoinRunner to load full job config from config loader when present.
URL: https://github.com/apache/samza/pull/1255#discussion_r372124889
##########
File path: samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
##########
@@ -111,29 +112,41 @@ public LocalApplicationRunner(SamzaApplication app, Config config) {
* @param metadataStoreFactory the instance of {@link MetadataStoreFactory} to read and write to coordinator stream.
*/
public LocalApplicationRunner(SamzaApplication app, Config config, MetadataStoreFactory metadataStoreFactory) {
- this(ApplicationDescriptorUtil.getAppDescriptor(app, config), getCoordinationUtils(config), metadataStoreFactory);
+ this(app, config, getCoordinationUtils(config), metadataStoreFactory);
}
/**
* Constructor only used in unit test to allow injection of {@link LocalJobPlanner}
*/
@VisibleForTesting
- LocalApplicationRunner(ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc, Optional<CoordinationUtils> coordinationUtils) {
- this(appDesc, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(appDesc.getConfig())));
+ LocalApplicationRunner(SamzaApplication app, Config config, Optional<CoordinationUtils> coordinationUtils) {
+ this(app, config, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(config)));
}
private LocalApplicationRunner(
- ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc,
+ SamzaApplication app,
+ Config config,
Optional<CoordinationUtils> coordinationUtils,
MetadataStoreFactory metadataStoreFactory) {
- this.appDesc = appDesc;
+ this.appDesc = getApplicationDescriptor(app, config);
this.isAppModeBatch = isAppModeBatch(appDesc.getConfig());
this.coordinationUtils = coordinationUtils;
this.metadataStoreFactory = Optional.ofNullable(metadataStoreFactory);
}
+ @VisibleForTesting
+ static ApplicationDescriptorImpl<? extends ApplicationDescriptor> getApplicationDescriptor(SamzaApplication app, Config config) {
+ return new JobConfig(config).getConfigLoaderFactory().isPresent()
+ ? ApplicationDescriptorUtil.getAppDescriptor(app, ConfigUtil.loadConfig(config))
+ : ApplicationDescriptorUtil.getAppDescriptor(app, config);
+ }
+
@VisibleForTesting
static MetadataStoreFactory getDefaultCoordinatorStreamStoreFactory(JobConfig jobConfig) {
+ if (jobConfig.getConfigLoaderFactory().isPresent()) {
+ jobConfig = new JobConfig(ConfigUtil.loadConfig(jobConfig));
+ }
Review comment:
Ideally yes, in order to achieve this, each constructor needs to have some if checks, thus it cannot invoke another constructor anymore. This will return in all constructors follows its own logic instead of merging into a single private constructor. Do you think this is a good trade off?
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] cameronlee314 merged pull request #1255: SAMZA-2429: Update
LocalApplicatoinRunner to load full job config from config loader when
present.
Posted by GitBox <gi...@apache.org>.
cameronlee314 merged pull request #1255: SAMZA-2429: Update LocalApplicatoinRunner to load full job config from config loader when present.
URL: https://github.com/apache/samza/pull/1255
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] kw2542 commented on a change in pull request #1255:
SAMZA-2429: Update LocalApplicatoinRunner to load full job config from
config loader when present.
Posted by GitBox <gi...@apache.org>.
kw2542 commented on a change in pull request #1255: SAMZA-2429: Update LocalApplicatoinRunner to load full job config from config loader when present.
URL: https://github.com/apache/samza/pull/1255#discussion_r372149172
##########
File path: samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
##########
@@ -150,6 +163,10 @@ static MetadataStoreFactory getDefaultCoordinatorStreamStoreFactory(JobConfig jo
}
private static Optional<CoordinationUtils> getCoordinationUtils(Config config) {
+ if (new JobConfig(config).getConfigLoaderFactory().isPresent()) {
+ config = ConfigUtil.loadConfig(config);
+ }
Review comment:
Updated
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] kw2542 commented on a change in pull request #1255:
SAMZA-2429: Update LocalApplicatoinRunner to load full job config from
config loader when present.
Posted by GitBox <gi...@apache.org>.
kw2542 commented on a change in pull request #1255: SAMZA-2429: Update LocalApplicatoinRunner to load full job config from config loader when present.
URL: https://github.com/apache/samza/pull/1255#discussion_r371577613
##########
File path: samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
##########
@@ -111,27 +112,35 @@ public LocalApplicationRunner(SamzaApplication app, Config config) {
* @param metadataStoreFactory the instance of {@link MetadataStoreFactory} to read and write to coordinator stream.
*/
public LocalApplicationRunner(SamzaApplication app, Config config, MetadataStoreFactory metadataStoreFactory) {
- this(ApplicationDescriptorUtil.getAppDescriptor(app, config), getCoordinationUtils(config), metadataStoreFactory);
+ this(app, config, getCoordinationUtils(config), metadataStoreFactory);
Review comment:
Very good question, and nice catch. You are right, both getCoordinationUtils as well as getDefaultCoordinatorStreamStoreFactory need to be updated to load full config first if applicable.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] kw2542 commented on a change in pull request #1255:
SAMZA-2429: Update LocalApplicatoinRunner to load full job config from
config loader when present.
Posted by GitBox <gi...@apache.org>.
kw2542 commented on a change in pull request #1255: SAMZA-2429: Update LocalApplicatoinRunner to load full job config from config loader when present.
URL: https://github.com/apache/samza/pull/1255#discussion_r371577749
##########
File path: samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
##########
@@ -111,27 +112,35 @@ public LocalApplicationRunner(SamzaApplication app, Config config) {
* @param metadataStoreFactory the instance of {@link MetadataStoreFactory} to read and write to coordinator stream.
*/
public LocalApplicationRunner(SamzaApplication app, Config config, MetadataStoreFactory metadataStoreFactory) {
- this(ApplicationDescriptorUtil.getAppDescriptor(app, config), getCoordinationUtils(config), metadataStoreFactory);
+ this(app, config, getCoordinationUtils(config), metadataStoreFactory);
}
/**
* Constructor only used in unit test to allow injection of {@link LocalJobPlanner}
*/
@VisibleForTesting
- LocalApplicationRunner(ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc, Optional<CoordinationUtils> coordinationUtils) {
- this(appDesc, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(appDesc.getConfig())));
+ LocalApplicationRunner(SamzaApplication app, Config config, Optional<CoordinationUtils> coordinationUtils) {
+ this(app, config, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(config)));
Review comment:
Updated.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] cameronlee314 commented on a change in pull request #1255:
SAMZA-2429: Update LocalApplicatoinRunner to load full job config from
config loader when present.
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #1255: SAMZA-2429: Update LocalApplicatoinRunner to load full job config from config loader when present.
URL: https://github.com/apache/samza/pull/1255#discussion_r371584196
##########
File path: samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
##########
@@ -111,27 +112,35 @@ public LocalApplicationRunner(SamzaApplication app, Config config) {
* @param metadataStoreFactory the instance of {@link MetadataStoreFactory} to read and write to coordinator stream.
*/
public LocalApplicationRunner(SamzaApplication app, Config config, MetadataStoreFactory metadataStoreFactory) {
- this(ApplicationDescriptorUtil.getAppDescriptor(app, config), getCoordinationUtils(config), metadataStoreFactory);
+ this(app, config, getCoordinationUtils(config), metadataStoreFactory);
}
/**
* Constructor only used in unit test to allow injection of {@link LocalJobPlanner}
*/
@VisibleForTesting
- LocalApplicationRunner(ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc, Optional<CoordinationUtils> coordinationUtils) {
- this(appDesc, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(appDesc.getConfig())));
+ LocalApplicationRunner(SamzaApplication app, Config config, Optional<CoordinationUtils> coordinationUtils) {
+ this(app, config, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(config)));
}
private LocalApplicationRunner(
Review comment:
In my opinion, `LocalApplicationRunner` doesn't truly belong in samza-core, but it had to be somewhere...
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] kw2542 commented on a change in pull request #1255:
SAMZA-2429: Update LocalApplicatoinRunner to load full job config from
config loader when present.
Posted by GitBox <gi...@apache.org>.
kw2542 commented on a change in pull request #1255: SAMZA-2429: Update LocalApplicatoinRunner to load full job config from config loader when present.
URL: https://github.com/apache/samza/pull/1255#discussion_r372139453
##########
File path: samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
##########
@@ -111,29 +112,41 @@ public LocalApplicationRunner(SamzaApplication app, Config config) {
* @param metadataStoreFactory the instance of {@link MetadataStoreFactory} to read and write to coordinator stream.
*/
public LocalApplicationRunner(SamzaApplication app, Config config, MetadataStoreFactory metadataStoreFactory) {
- this(ApplicationDescriptorUtil.getAppDescriptor(app, config), getCoordinationUtils(config), metadataStoreFactory);
+ this(app, config, getCoordinationUtils(config), metadataStoreFactory);
}
/**
* Constructor only used in unit test to allow injection of {@link LocalJobPlanner}
*/
@VisibleForTesting
- LocalApplicationRunner(ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc, Optional<CoordinationUtils> coordinationUtils) {
- this(appDesc, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(appDesc.getConfig())));
+ LocalApplicationRunner(SamzaApplication app, Config config, Optional<CoordinationUtils> coordinationUtils) {
+ this(app, config, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(config)));
}
private LocalApplicationRunner(
- ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc,
+ SamzaApplication app,
+ Config config,
Optional<CoordinationUtils> coordinationUtils,
MetadataStoreFactory metadataStoreFactory) {
- this.appDesc = appDesc;
+ this.appDesc = getApplicationDescriptor(app, config);
this.isAppModeBatch = isAppModeBatch(appDesc.getConfig());
this.coordinationUtils = coordinationUtils;
this.metadataStoreFactory = Optional.ofNullable(metadataStoreFactory);
}
+ @VisibleForTesting
+ static ApplicationDescriptorImpl<? extends ApplicationDescriptor> getApplicationDescriptor(SamzaApplication app, Config config) {
+ return new JobConfig(config).getConfigLoaderFactory().isPresent()
+ ? ApplicationDescriptorUtil.getAppDescriptor(app, ConfigUtil.loadConfig(config))
+ : ApplicationDescriptorUtil.getAppDescriptor(app, config);
+ }
+
@VisibleForTesting
static MetadataStoreFactory getDefaultCoordinatorStreamStoreFactory(JobConfig jobConfig) {
+ if (jobConfig.getConfigLoaderFactory().isPresent()) {
+ jobConfig = new JobConfig(ConfigUtil.loadConfig(jobConfig));
+ }
Review comment:
I like the idea of LocalApplicationRunnerContext, will update the PR.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] kw2542 commented on a change in pull request #1255:
SAMZA-2429: Update LocalApplicatoinRunner to load full job config from
config loader when present.
Posted by GitBox <gi...@apache.org>.
kw2542 commented on a change in pull request #1255: SAMZA-2429: Update LocalApplicatoinRunner to load full job config from config loader when present.
URL: https://github.com/apache/samza/pull/1255#discussion_r371578346
##########
File path: samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
##########
@@ -111,27 +112,35 @@ public LocalApplicationRunner(SamzaApplication app, Config config) {
* @param metadataStoreFactory the instance of {@link MetadataStoreFactory} to read and write to coordinator stream.
*/
public LocalApplicationRunner(SamzaApplication app, Config config, MetadataStoreFactory metadataStoreFactory) {
- this(ApplicationDescriptorUtil.getAppDescriptor(app, config), getCoordinationUtils(config), metadataStoreFactory);
+ this(app, config, getCoordinationUtils(config), metadataStoreFactory);
}
/**
* Constructor only used in unit test to allow injection of {@link LocalJobPlanner}
*/
@VisibleForTesting
- LocalApplicationRunner(ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc, Optional<CoordinationUtils> coordinationUtils) {
- this(appDesc, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(appDesc.getConfig())));
+ LocalApplicationRunner(SamzaApplication app, Config config, Optional<CoordinationUtils> coordinationUtils) {
+ this(app, config, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(config)));
}
private LocalApplicationRunner(
Review comment:
Good question, I would consider changes in samza-api module to be public facing API changes, and changes in samza-core to be internal API changes, which can be less risky.
I will update the PR description to explicitly call out the constructor changes, also this is a backward compatible API change.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] cameronlee314 commented on a change in pull request #1255:
SAMZA-2429: Update LocalApplicatoinRunner to load full job config from
config loader when present.
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #1255: SAMZA-2429: Update LocalApplicatoinRunner to load full job config from config loader when present.
URL: https://github.com/apache/samza/pull/1255#discussion_r371563040
##########
File path: samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
##########
@@ -111,27 +112,35 @@ public LocalApplicationRunner(SamzaApplication app, Config config) {
* @param metadataStoreFactory the instance of {@link MetadataStoreFactory} to read and write to coordinator stream.
*/
public LocalApplicationRunner(SamzaApplication app, Config config, MetadataStoreFactory metadataStoreFactory) {
- this(ApplicationDescriptorUtil.getAppDescriptor(app, config), getCoordinationUtils(config), metadataStoreFactory);
+ this(app, config, getCoordinationUtils(config), metadataStoreFactory);
}
/**
* Constructor only used in unit test to allow injection of {@link LocalJobPlanner}
*/
@VisibleForTesting
- LocalApplicationRunner(ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc, Optional<CoordinationUtils> coordinationUtils) {
- this(appDesc, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(appDesc.getConfig())));
+ LocalApplicationRunner(SamzaApplication app, Config config, Optional<CoordinationUtils> coordinationUtils) {
+ this(app, config, coordinationUtils, getDefaultCoordinatorStreamStoreFactory(new JobConfig(config)));
}
private LocalApplicationRunner(
Review comment:
Is this considered a semantic API change? Before, it seems like `config` was always the full config, but now you are allowing the initial config with only config loader properties.
It's ok to have this be an API change, but just wanted to make sure you note that where necessary.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services