Sanil15 commented on a change in pull request #917: Adding bounded-time start
of SamzaContainer from StreamProcessor
URL: https://github.com/apache/samza/pull/917#discussion_r256664270
##########
File path:
samza-core/src/main/java/org/apache/samza/processor/StreamProcessor.java
##########
@@ -401,11 +403,29 @@ public void onJobModelExpired() {
public void onNewJobModel(String processorId, JobModel jobModel) {
synchronized (lock) {
if (state == State.IN_REBALANCE) {
+ containerStartLatch = new CountDownLatch(1);
containerShutdownLatch = new CountDownLatch(1);
container = createSamzaContainer(processorId, jobModel);
container.setContainerListener(new ContainerListener());
LOGGER.info("Starting the container: {} for the stream processor:
{}.", container, processorId);
containerExcecutorService.submit(container);
+
+ try {
+ LOGGER.info("Waiting {} ms for the container: {} to startup.",
taskStartMs, container);
+
+ if (!containerStartLatch.await(taskStartMs,
TimeUnit.MILLISECONDS)) {
+ LOGGER.error("Timed out {} ms while waiting for the container:
{} to startup. Interrupting the container: {} thread to die", taskStartMs,
container, container);
+ containerExcecutorService.shutdownNow();
+ state = State.STOPPED;
+ jobCoordinator.stop();
+ processorListener.afterFailure(new TimeoutException("Container
start timed out after "+taskStartMs+ " ms"));
+ }
+ } catch (InterruptedException e) {
+ LOGGER.error("Exception occurred when starting the container:
{}.", container, e);
+ state = State.STOPPED;
+ jobCoordinator.stop();
Review comment:
Shouldn't this be in finally?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on 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
