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) {
   
    final List<ListeningExecutorService> l =
        Spez.ServicePoolGenerator(THREAD_POOL, "Spanner Tailer Event Worker");
    final SpannerTailer tailer = new SpannerTailer(THREAD_POOL, 200000000);
   
    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);
   
    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) { final list<listeningexecutorservice> l = spez.servicepoolgenerator(thread_pool, "spanner tailer event worker"); final spannertailer tailer = new spannertailer(thread_pool, 200000000); 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); 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(); } }