class FullTraversalConnector implements IndexingConnector, IncrementalChangeHandler {

  @Override
  public void init(IndexingConnectorContext context) throws Exception {
    checkState(Configuration.isInitialized(), "configuration not initialized");
    indexingService = checkNotNull(context.getIndexingService());
    defaultAcl = DefaultAcl.fromConfiguration(indexingService);
    repositoryContext =
        new RepositoryContext.Builder()
            .setEventBus(new EventBus("EventBus-" + getClass().getName()))
            .setDefaultAclMode(defaultAcl.getDefaultAclMode())
            .build();
    if (checkpointHandler == null) {
      checkpointHandler = LocalFileCheckpointHandler.fromConfiguration();
    }

    useQueues = Configuration.getBoolean(TRAVERSE_USE_QUEUES, DEFAULT_USE_QUEUES).get();
    QueueCheckpoint.Builder builder = new QueueCheckpoint.Builder(useQueues);
    if (useQueues) {
      String tag =
          Configuration.getString(TRAVERSE_QUEUE_TAG, repository.getClass().getSimpleName()).get();
      builder.setCheckpointHandler(checkpointHandler)
          .setQueueA(QUEUE_NAME + "A: " + tag)
          .setQueueB(QUEUE_NAME + "B: " + tag);
    }
    queueCheckpoint = builder.build();

    numThreads = Configuration.getInteger(NUM_THREADS, DEFAULT_THREAD_NUM).get();
    // TODO(bmj): Fix this gross violation of encapsulation.
    numToAbort = Configuration
        .getValue(TraverseExceptionHandlerFactory.TRAVERSE_EXCEPTION_HANDLER, 0L, value -> {
          if (IGNORE_FAILURE.equals(value)) {
            return Long.MAX_VALUE;
          } else {
            try {
              return Long.parseLong(value);
            } catch (NumberFormatException e) {
              throw new InvalidConfigurationException(
                  "Unrecognized value for traversal exception handler: " + value, e);
            }
          }
        }).get();

    threadPoolExecutor =
        new ThreadPoolExecutor(
            numThreads,
            numThreads,
            0,
            TimeUnit.MILLISECONDS,
            new ArrayBlockingQueue<Runnable>(10 * numThreads),
            new ThreadPoolExecutor.CallerRunsPolicy());
    listeningExecutorService = MoreExecutors.listeningDecorator(threadPoolExecutor);
    partitionSize = Configuration.getInteger(TRAVERSE_PARTITION_SIZE, DEFAULT_PARTITION_SIZE).get();
    Configuration.checkConfiguration(
        partitionSize > 0,
        "Partition size can not be less than or equal to 0. Configured value %s",
        partitionSize);
    repositoryContext.getEventBus().register(this);
    repository.init(repositoryContext);
    logger.log(Level.INFO, "start full traversal connector executors");
  }

  public  FullTraversalConnector(Repository repository);
  public  FullTraversalConnector(Repository repository, CheckpointHandler checkpointHandler);

}