private void window() {
      state.startWindow();
      Runnable windowWorker = new Runnable() {
        @Override
        public void run() {
          try {
            containerMetrics.windows().inc();

            ReadableCoordinator coordinator = new ReadableCoordinator(task.taskName());
            long startTime = clock.nanoTime();
            task.window(coordinator);
            containerMetrics.windowNs().update(clock.nanoTime() - startTime);

            /**
             * Window calls that execute for more than task.window.ms will starve process calls
             * since window has higher priority than process in {@link AsyncTaskState#nextOp()}.
             * Warn the users if this is the case.
             */
            long averageWindowMs = TimeUnit.NANOSECONDS.toMillis(
                (long) containerMetrics.windowNs().getSnapshot().getAverage());
            if (averageWindowMs >= windowMs) {
              log.warn("Average window call duration {} is greater than the configured task.window.ms {}. " +
                      "This can starve process calls, so consider setting task.window.ms >> {} ms.",
                  new Object[]{averageWindowMs, windowMs, averageWindowMs});
            }

            coordinatorRequests.update(coordinator);

            state.doneWindow();
          } catch (Throwable t) {
            log.error("Task {} window failed", task.taskName(), t);
            abort(t);
          } finally {
            log.trace("Task {} window completed", task.taskName());
            resume();
          }
        }
      };

      if (threadPool != null) {
        log.trace("Task {} window on the thread pool", task.taskName());
        threadPool.submit(windowWorker);
      } else {
        log.trace("Task {} window on the run loop thread", task.taskName());
        windowWorker.run();
      }
    }
Invoke window. Run window in thread pool if not the single thread mode.
samza-core/src/main/java/org/apache/samza/container/RunLoop.java