SAMZA-1832: Fix race condition between StreamProcessor and SamzaContainerListener

[bharathkk]

The PR addresses the following issues

• We have few scenarios where there is a race condition between SamzaContainerListener and StreamProcessor that results in incorrect application status being propagated to client

Consider the scenario when the container runs into an exception in run loop and triggers shutdown sequence and at the same time the user triggers a stop on the StreamProcessor. The user request comes in and notices that the container is already shutting down and proceeds to shutdown the job coordinator which in turns shuts down successfully and the processorListener.onStop() is invoked. The container eventually invokes the SamzaContainerListener callback and updates the exception state after the StreamProcessor has finished shutting down.

• Currently, we only propagate failures to processorListener when containerException is not null. It is possible for the samza container to take longer than task.shutdown.ms to shutdown in which case, we need to propagate a timeout exception to the processorListener as opposed assuming the shutdown was successful.

• Make container shutdown idempotent since its only setting a boolean flag to true and there is no need to throw an exception on subsequent invocations.
  1. It simplifies the interaction of other components (StreamProcessor) with container and prevents unnecessary check on container state to determine if its safe to call shutdown or not.
  2. Enables to reason about the impact of container state change on StreamProcessor easily since we restrict container interaction w/ StreamProcessor only via SamzaContainerListener.

[bharathkk]

@nickpan47 @prateekm @shanthoosh
Please take a look

[shanthoosh]

Looks like there're compilation errors in new tests, please fix them.

[nickpan47]

minor comment. lgtm. Thanks!