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_r265688727
##########
File path:
samza-core/src/main/java/org/apache/samza/execution/LocalJobPlanner.java
##########
@@ -119,7 +124,17 @@ private void createStreams(String planId,
List<StreamSpec> intStreams, StreamMan
return;
}
- DistributedLockWithState lockWithState =
coordinationUtils.getLockWithState(planId);
+ // If BATCH, then need to create new intermediate streams every run.
+ // planId does not change every run and hence, need to use runid
+ // as the lockId to create a new lock with state each run
+ // to create new streams each run.
+ // If run.id is null, defaults to old behavior of using planId
+ Boolean isAppModeBatch = new ApplicationConfig(userConfig).getAppMode() ==
ApplicationConfig.ApplicationMode.BATCH;
+ String lockId = planId;
+ if(isAppModeBatch && runId != null) {
Review comment:
UnsupportedOperation is thrown in LocalApplicationRunner and not in
LocalJobPlanner. Thus there is no way to know why runId is null and hence
keeping this as is.
----------------------------------------------------------------
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
