This is a weird api to expose to subclasses. How would an implr know when it is safe to invoke this?

The exception is being re-thrown anyway, so why not handle it in the super class? I don't like how tightly the super and sub classes are being coupled here.

@vmaheshw @surajkn @ryannedolan

Let me take this up in a separate PR but to reduce the coupling here, could we do something like this in the AbstractKafkaConnector(from where the task threads are initiated).

  private ConnectorTaskEntry createKafkaConnectorTask(DatastreamTask task) {
    _logger.info("Creating connector task for datastream task {}.", task);
    AbstractKafkaBasedConnectorTask connectorTask = createKafkaBasedConnectorTask(task);
    Thread taskThread = createTaskThread(connectorTask);
    taskThread.start();

    // task cleanup thread cleans up the task resources like stopped latch counts post shutdown of the task
    Thread taskCleanUpThread = new Thread(() -> {
      try {
        taskThread.join();
      } catch (InterruptedException exception) {
        _logger.error(String.format("Got interrupted exception while waiting for the completion of task : %s ", connectorTask), exception);
      } finally {
        connectorTask.postShutdownCleanUp();
      }
    });
    taskCleanUpThread.start();
    return new ConnectorTaskEntry(connectorTask, taskThread);
  } 

the postShutdownCleanUp method here would take care of counting the latch down so we don't have to handle it in run() functions of either derived or base class implementations.

Basically it will create extra thread for each thread. I think it should work, but with the constraint of double task threads.

@vmaheshw I remember you had expressed concern over this way of doing the count down. IIRC your concern was that any subclass of this can forget to call countDownStopppedLatch (although in this patch Shrinand added a call to it in KafkaMirrorMakerConnectorTask). So just wanted to check if that was considered and concluded that this is the most reasonable way of doing it?
cc @shrinandthakkar

Yes, I was not in support of doing it this way, but also did not have a better idea. So, we decided to do it, until we have a better idea.