vjagadish1989 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_r286722325
##########
File path:
samza-core/src/main/java/org/apache/samza/execution/LocalJobPlanner.java
##########
@@ -104,36 +127,89 @@ private void createStreams(String planId,
List<StreamSpec> intStreams, StreamMan
LOG.info("Set of intermediate streams is empty. Nothing to create.");
return;
}
- LOG.info("A single processor must create the intermediate streams.
Processor {} will attempt to acquire the lock.", uid);
+ LOG.info("A single processor must create the intermediate streams.
Processor {} will attempt to acquire the lock.", processorId);
// Move the scope of coordination utils within stream creation to address
long idle connection problem.
// Refer SAMZA-1385 for more details
- JobCoordinatorConfig jcConfig = new JobCoordinatorConfig(userConfig);
- String coordinationId = new ApplicationConfig(userConfig).getGlobalAppId()
+ APPLICATION_RUNNER_PATH_SUFFIX;
- CoordinationUtils coordinationUtils =
-
jcConfig.getCoordinationUtilsFactory().getCoordinationUtils(coordinationId,
uid, userConfig);
if (coordinationUtils == null) {
- LOG.warn("Processor {} failed to create utils. Each processor will
attempt to create streams.", uid);
+ LOG.warn("Processor {} failed to create utils. Each processor will
attempt to create streams.", processorId);
// each application process will try creating the streams, which
// requires stream creation to be idempotent
streamManager.createStreams(intStreams);
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) {
+ lockId = runId;
+ }
try {
- // check if the processor needs to go through leader election and stream
creation
- if (lockWithState.lockIfNotSet(1000, TimeUnit.MILLISECONDS)) {
- LOG.info("lock acquired for streams creation by " + uid);
- streamManager.createStreams(intStreams);
- lockWithState.unlockAndSet();
- } else {
- LOG.info("Processor {} did not obtain the lock for streams creation.
They must've been created by another processor.", uid);
- }
- } catch (TimeoutException e) {
- String msg = String.format("Processor {} failed to get the lock for
stream initialization", uid);
- throw new SamzaException(msg, e);
+ checkAndCreateStreams(lockId, intStreams, streamManager);
+ } catch (TimeoutException te) {
+ throw new SamzaException(String.format("Processor {} failed to get the
lock for stream initialization within timeout.", processorId), te);
} finally {
- coordinationUtils.close();
+ if (!isAppModeBatch && coordinationUtils != null) {
+ coordinationUtils.close();
+ }
+ }
+ }
+
+ private void checkAndCreateStreams(String lockId, List<StreamSpec>
intStreams, StreamManager streamManager) throws TimeoutException {
+ MetadataStore metadataStore = getMetadataStore();
+ DistributedLock distributedLock = coordinationUtils.getLock(lockId);
Review comment:
pass in the lock as a param to this 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.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
