class EventWatcher {

  public void start(EventQueue queue) {

    queue.addHandler(startEvent, this::handleStartEvent);

    queue.addHandler(completionEvent, (completionPayload, publisher) -> {
      try {
        handleCompletion(completionPayload, publisher);
      } catch (LeaseNotFoundException e) {
        //It's already done. Too late, we already retried it.
        //Not a real problem.
        LOG.warn("Completion event without active lease. Id={}",
            completionPayload.getId());
      }
    });

    MetricsSystem ms = DefaultMetricsSystem.instance();
    ms.register(name, "EventWatcher metrics", metrics);
  }

  public  EventWatcher(String name, Event<TIMEOUT_PAYLOAD> startEvent,
      Event<COMPLETION_PAYLOAD> completionEvent,
      LeaseManager<Long> leaseManager);
  public  EventWatcher(Event<TIMEOUT_PAYLOAD> startEvent,
      Event<COMPLETION_PAYLOAD> completionEvent,
      LeaseManager<Long> leaseManager);

}