@Override
    public void onComplete(final TaskCallback callback) {
      long workNanos = clock.nanoTime() - ((TaskCallbackImpl) callback).getTimeCreatedNs();
      callbackExecutor.schedule(new Runnable() {
        @Override
        public void run() {
          try {
            state.doneProcess();
            state.taskMetrics.asyncCallbackCompleted().inc();
            TaskCallbackImpl callbackImpl = (TaskCallbackImpl) callback;
            containerMetrics.processNs().update(clock.nanoTime() - callbackImpl.getTimeCreatedNs());
            log.trace("Got callback complete for task {}, ssp {}",
                callbackImpl.getTaskName(), callbackImpl.getSystemStreamPartition());

            List<TaskCallbackImpl> callbacksToUpdate = callbackManager.updateCallback(callbackImpl);
            for (TaskCallbackImpl callbackToUpdate : callbacksToUpdate) {
              log.trace("Update offset for ssp {}, offset {}", callbackToUpdate.getSystemStreamPartition(),
                  callbackToUpdate.getOffset());

              // update offset
              if (task.offsetManager() != null) {
                task.offsetManager().update(task.taskName(), callbackToUpdate.getSystemStreamPartition(),
                    callbackToUpdate.getOffset());
              }

              // update coordinator
              coordinatorRequests.update(callbackToUpdate.getCoordinator());
            }
          } catch (Throwable t) {
            log.error("Error marking process as complete.", t);
            abort(t);
          } finally {
            resume();
          }
        }
      }, workNanos);
    }
Task process completes successfully, update the offsets based on the high-water mark.
Then it will trigger the listener for task state change.
* @param callback AsyncSteamTask.processAsync callback
samza-core/src/main/java/org/apache/samza/container/RunLoop.java