lakshmi-manasa-g commented on a change in pull request #938: SAMZA-1531:
Support run.id in standalone for batch processing.
URL: https://github.com/apache/samza/pull/938#discussion_r264493684
##########
File path:
samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
##########
@@ -77,23 +95,80 @@
*/
public LocalApplicationRunner(SamzaApplication app, Config config) {
this.appDesc = ApplicationDescriptorUtil.getAppDescriptor(app, config);
- this.planner = new LocalJobPlanner(appDesc);
+ this.coordinationUtils = getCoordinationUtils(config);
+ getRunId();
+ this.planner = new LocalJobPlanner(appDesc, coordinationUtils, uid, runId);
}
/**
* Constructor only used in unit test to allow injection of {@link
LocalJobPlanner}
*/
@VisibleForTesting
- LocalApplicationRunner(ApplicationDescriptorImpl<? extends
ApplicationDescriptor> appDesc, LocalJobPlanner planner) {
+ LocalApplicationRunner(ApplicationDescriptorImpl<? extends
ApplicationDescriptor> appDesc, LocalJobPlanner planner, CoordinationUtils
coordinationUtils) {
this.appDesc = appDesc;
this.planner = planner;
+ this.coordinationUtils = coordinationUtils;
+ }
+
+ private CoordinationUtils getCoordinationUtils(Config config) {
+ JobCoordinatorConfig jcConfig = new JobCoordinatorConfig(config);
+ String coordinationId = new ApplicationConfig(config).getGlobalAppId() +
APPLICATION_RUNNER_PATH_SUFFIX;
+ return
jcConfig.getCoordinationUtilsFactory().getCoordinationUtils(coordinationId,
uid, config);
+ }
+
+ private void getRunId(){
+ Boolean isAppModeBatch = new
ApplicationConfig(appDesc.getConfig()).getAppMode() ==
ApplicationConfig.ApplicationMode.BATCH;
+ if(coordinationUtils == null || !isAppModeBatch) {
+ return;
+ }
+
+ runIdLock = coordinationUtils.getReadWriteLock(RUNID_LOCK_ID);
+ if(runIdLock == null) {
+ LOG.warn("Processor {} failed to create the lock for run.id generation",
uid);
Review comment:
Actually, I am using uid is several Log statements. pl let me know if i
should remove it out of all of them.
here there are two cases runIdLock could be null -- ZK failure which will
anways throw an exception trying to create ZkDistributedReadWriteLock or in
Azure where it is not supported. For Azure, we really dont have any other info
to provide.
Would it help to provide an elaborate warning -- something like runid lock
creation failed and that old run's state (intermediate streams etc) will be
used?
----------------------------------------------------------------
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:
us...@infra.apache.org
With regards,
Apache Git Services
