public void close() {
    if (!closed && processorId != null) {
      closed = true;
      clusterMembership.unregisterProcessor(processorId);
    }
  }
might be called several times and hence should be idempotent
samza-core/src/main/java/org/apache/samza/coordinator/RunIdGenerator.java