// TODO(pdex): why are we making our own threadpool?
public static void main(String[] args) {
    // TODO(pdex): why are we making our own threadpool?
    final List<ListeningExecutorService> l =
        Spez.ServicePoolGenerator(THREAD_POOL, "Spanner Tailer Event Worker");
    final SpannerTailer tailer = new SpannerTailer(THREAD_POOL, 200000000);
    // final EventPublisher publisher = new EventPublisher(PROJECT_NAME, TOPIC_NAME);
    final ThreadLocal<EventPublisher> publisher =
        ThreadLocal.withInitial(
            () -> {
              return new EventPublisher(PROJECT_NAME, TOPIC_NAME);
            });
    final ExecutorService workStealingPool = Executors.newWorkStealingPool();
    final ListeningExecutorService forkJoinPool =
        MoreExecutors.listeningDecorator(workStealingPool);
    final Map<String, String> metadata = new HashMap<>();
    final CountDownLatch doneSignal = new CountDownLatch(1);
    final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
    // Populate CDC Metadata
    metadata.put("SrcDatabase", DB_NAME);
    metadata.put("SrcTablename", TABLE_NAME);
    metadata.put("DstTopic", TOPIC_NAME);
    final ListenableFuture<SchemaSet> schemaSetFuture =
        tailer.getSchema(PROJECT_NAME, INSTANCE_NAME, DB_NAME, TABLE_NAME);
    Futures.addCallback(
        schemaSetFuture,
        new FutureCallback<SchemaSet>() {
          @Override
          public void onSuccess(SchemaSet schemaSet) {
            log.info("Successfully Processed the Table Schema. Starting the poller now ...");
            if (DISRUPTOR) {
              DisruptorHandler handler =
                  new DisruptorHandler(schemaSet, publisher, metadata, l.get(0));
              handler.start();
              tailer.setRingBuffer(handler.getRingBuffer());
              ScheduledFuture<?> result =
                  tailer.start(
                      2,
                      500,
                      PROJECT_NAME,
                      INSTANCE_NAME,
                      DB_NAME,
                      TABLE_NAME,
                      "lpts_table",
                      schemaSet.tsColName(),
                      "2000");
              doneSignal.countDown();
            } else {
              WorkStealingHandler handler =
                  new WorkStealingHandler(scheduler, schemaSet, publisher, metadata);
              tailer.start(
                  handler,
                  schemaSet.tsColName(),
                  l.size(),
                  THREAD_POOL,
                  500,
                  PROJECT_NAME,
                  INSTANCE_NAME,
                  DB_NAME,
                  TABLE_NAME,
                  "lpts_table",
                  "2000",
                  500,
                  500);
              scheduler.scheduleAtFixedRate(
                  () -> {
                    handler.logStats();
                    tailer.logStats();
                  },
                  30,
                  30,
                  TimeUnit.SECONDS);
              doneSignal.countDown();
            }
          }
          @Override
          public void onFailure(Throwable t) {
            log.error("Unable to process schema", t);
            System.exit(-1);
          }
        },
        l.get(l.size() % THREAD_POOL));
    try {
      log.debug("Dumping all known Loggers");
      LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
      java.util.Iterator<ch.qos.logback.classic.Logger> it = lc.getLoggerList().iterator();
      while (it.hasNext()) {
        ch.qos.logback.classic.Logger thisLog = it.next();
        log.debug("name: {} status: {}", thisLog.getName(), thisLog.getLevel());
      }
      log.info("waiting for doneSignal");
      doneSignal.await();
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
public static void main(String[] args) {
// TODO(pdex): why are we making our own threadpool?
final List<ListeningExecutorService> l =
Spez.ServicePoolGenerator(THREAD_POOL, "Spanner Tailer Event Worker");
public static void main(String[] args) {
// TODO(pdex): why are we making our own threadpool?
final List<ListeningExecutorService> l =
Spez.ServicePoolGenerator(THREAD_POOL, "Spanner Tailer Event Worker");
final SpannerTailer tailer = new SpannerTailer(THREAD_POOL, 200000000);
// final EventPublisher publisher = new EventPublisher(PROJECT_NAME, TOPIC_NAME);
final ThreadLocal<EventPublisher> publisher =
ThreadLocal.withInitial(
() -> {
return new EventPublisher(PROJECT_NAME, TOPIC_NAME);
});
final ExecutorService workStealingPool = Executors.newWorkStealingPool();
public static void main(String[] args) {
// TODO(pdex): why are we making our own threadpool?
final List<ListeningExecutorService> l =
Spez.ServicePoolGenerator(THREAD_POOL, "Spanner Tailer Event Worker");
final SpannerTailer tailer = new SpannerTailer(THREAD_POOL, 200000000);
// final EventPublisher publisher = new EventPublisher(PROJECT_NAME, TOPIC_NAME);
final ThreadLocal<EventPublisher> publisher =
ThreadLocal.withInitial(
() -> {
return new EventPublisher(PROJECT_NAME, TOPIC_NAME);
});
final ExecutorService workStealingPool = Executors.newWorkStealingPool();
final ListeningExecutorService forkJoinPool =
MoreExecutors.listeningDecorator(workStealingPool);
final Map<String, String> metadata = new HashMap<>();
final CountDownLatch doneSignal = new CountDownLatch(1);
final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
// Populate CDC Metadata
metadata.put("SrcDatabase", DB_NAME);
metadata.put("SrcTablename", TABLE_NAME);
metadata.put("DstTopic", TOPIC_NAME);
final ListenableFuture<SchemaSet> schemaSetFuture =