@Test
public void testGetErrorMultipleResources() throws Exception {
  TerrapinInternalGetRequest request = new TerrapinInternalGetRequest();
  MultiKey multiKey1 = new MultiKey().setResource("resource1").setPartition("1");
  multiKey1.addToKey(ByteBuffer.wrap("k1".getBytes()));
  MultiKey multiKey2 = new MultiKey().setResource("resource2").setPartition("1");
  multiKey2.addToKey(ByteBuffer.wrap("k2".getBytes()));
  request.addToKeyList(multiKey1);
  request.addToKeyList(multiKey2);

  Reader mockReader = mock(Reader.class);
  when(mockResourcePartitionMap.getReader(eq("resource1"), eq("1"))).thenReturn(mockReader);
  Try<TerrapinResponse> responseTry = serverImpl.get(request).get(Duration.forever());
  TerrapinGetException e = (TerrapinGetException)((Throw)responseTry).e();
  assertEquals(TerrapinGetErrorCode.INVALID_REQUEST, e.getErrorCode());
}
 

public void startAdminHttpService() {
  try {
    Properties properties = new Properties();
    properties.load(this.getClass().getResource("build.properties").openStream());
    LOG.info("build.properties build_revision: {}",
        properties.getProperty("build_revision", "unknown"));
  } catch (Throwable t) {
    LOG.warn("Failed to load properties from build.properties", t);
  }
  Duration[] defaultLatchIntervals = {Duration.apply(1, TimeUnit.MINUTES)};
  Iterator<Duration> durationIterator = Arrays.asList(defaultLatchIntervals).iterator();
  @SuppressWarnings("deprecation")
  AdminServiceFactory adminServiceFactory = new AdminServiceFactory(
      this.port,
      20,
      List$.MODULE$.empty(),
      Option.empty(),
      List$.MODULE$.empty(),
      Map$.MODULE$.empty(),
      JavaConversions.asScalaIterator(durationIterator).toList());
  RuntimeEnvironment runtimeEnvironment = new RuntimeEnvironment(this);
  AdminHttpService service = adminServiceFactory.apply(runtimeEnvironment);
  for (Map.Entry<String, CustomHttpHandler> entry : this.customHttpHandlerMap.entrySet()) {
    service.httpServer().createContext(entry.getKey(), entry.getValue());
  }
}
 

public LeastLoadPlacementPolicy(LoadAppraiser loadAppraiser, RoutingService routingService,
                                Namespace namespace, PlacementStateManager placementStateManager,
                                Duration refreshInterval, StatsLogger statsLogger) {
    super(loadAppraiser, routingService, namespace, placementStateManager, refreshInterval, statsLogger);
    statsLogger.registerGauge("placement/load.diff", new Gauge<Number>() {
        @Override
        public Number getDefaultValue() {
            return 0;
        }

        @Override
        public Number getSample() {
            if (serverLoads.size() > 0) {
                return serverLoads.last().getLoad() - serverLoads.first().getLoad();
            } else {
                return getDefaultValue();
            }
        }
    });
}
 

/**
 * Sanity check to make sure both checksum flag values work.
 */
@Test(timeout = 60000)
public void testChecksumFlag() throws Exception {
    String name = "testChecksumFlag";
    LocalRoutingService routingService = LocalRoutingService.newBuilder().build();
    routingService.addHost(name, dlServer.getAddress());
    DistributedLogClientBuilder dlClientBuilder = DistributedLogClientBuilder.newBuilder()
        .name(name)
        .clientId(ClientId$.MODULE$.apply("test"))
        .routingService(routingService)
        .handshakeWithClientInfo(true)
        .clientBuilder(ClientBuilder.get()
            .hostConnectionLimit(1)
            .connectionTimeout(Duration.fromSeconds(1))
            .requestTimeout(Duration.fromSeconds(60)))
        .checksum(false);
    DistributedLogClient dlClient = dlClientBuilder.build();
    Await.result(dlClient.write(name, ByteBuffer.wrap(("1").getBytes())));
    dlClient.close();

    dlClient = dlClientBuilder.checksum(true).build();
    Await.result(dlClient.write(name, ByteBuffer.wrap(("2").getBytes())));
    dlClient.close();
}
 

@Test(timeout = 60000)
public void testBulkWriteEmptyBuffer() throws Exception {
    String name = String.format("dlserver-bulk-write-%s", "empty");

    dlClient.routingService.addHost(name, dlServer.getAddress());

    List<ByteBuffer> writes = new ArrayList<ByteBuffer>();
    writes.add(ByteBuffer.wrap(("").getBytes()));
    writes.add(ByteBuffer.wrap(("").getBytes()));
    List<Future<DLSN>> futures = dlClient.dlClient.writeBulk(name, writes);
    assertEquals(2, futures.size());
    for (Future<DLSN> future : futures) {
        // No throw == pass
        DLSN dlsn = Await.result(future, Duration.fromSeconds(10));
    }
}
 

@Test(timeout = 10000)
public void testCalculateBalances() throws Exception {
    int numSevers = new Random().nextInt(20) + 1;
    int numStreams = new Random().nextInt(200) + 1;
    RoutingService mockRoutingService = mock(RoutingService.class);
    Namespace mockNamespace = mock(Namespace.class);
    LeastLoadPlacementPolicy leastLoadPlacementPolicy = new LeastLoadPlacementPolicy(
        new EqualLoadAppraiser(),
        mockRoutingService,
        mockNamespace,
        null,
        Duration.fromSeconds(600),
        new NullStatsLogger());
    TreeSet<ServerLoad> serverLoads =
        Await.result(leastLoadPlacementPolicy.calculate(generateServers(numSevers), generateStreams(numStreams)));
    long lowLoadPerServer = numStreams / numSevers;
    long highLoadPerServer = lowLoadPerServer + 1;
    for (ServerLoad serverLoad : serverLoads) {
        long load = serverLoad.getLoad();
        assertEquals(load, serverLoad.getStreamLoads().size());
        assertTrue(String.format("Load %d is not between %d and %d",
            load, lowLoadPerServer, highLoadPerServer), load == lowLoadPerServer || load == highLoadPerServer);
    }
}
 

protected DLClient(String name,
                   String streamNameRegex,
                   Optional<String> serverSideRoutingFinagleName) {
    routingService = LocalRoutingService.newBuilder().build();
    dlClientBuilder = DistributedLogClientBuilder.newBuilder()
                .name(name)
                .clientId(ClientId$.MODULE$.apply(name))
                .routingService(routingService)
                .streamNameRegex(streamNameRegex)
                .handshakeWithClientInfo(true)
                .clientBuilder(ClientBuilder.get()
                    .hostConnectionLimit(1)
                    .connectionTimeout(Duration.fromSeconds(1))
                    .requestTimeout(Duration.fromSeconds(60)));
    if (serverSideRoutingFinagleName.isPresent()) {
        dlClientBuilder =
                dlClientBuilder.serverRoutingServiceFinagleNameStr(serverSideRoutingFinagleName.get());
    }
    dlClient = (DistributedLogClientImpl) dlClientBuilder.build();
}
 

protected TwoRegionDLClient(String name, Map<SocketAddress, String> regionMap) {
    localRoutingService = new LocalRoutingService();
    remoteRoutingService = new LocalRoutingService();
    RegionsRoutingService regionsRoutingService =
            RegionsRoutingService.of(new DefaultRegionResolver(regionMap),
                    localRoutingService, remoteRoutingService);
    dlClientBuilder = DistributedLogClientBuilder.newBuilder()
                .name(name)
                .clientId(ClientId$.MODULE$.apply(name))
                .routingService(regionsRoutingService)
                .streamNameRegex(".*")
                .handshakeWithClientInfo(true)
                .maxRedirects(2)
                .clientBuilder(ClientBuilder.get()
                    .hostConnectionLimit(1)
                    .connectionTimeout(Duration.fromSeconds(1))
                    .requestTimeout(Duration.fromSeconds(10)));
    dlClient = (DistributedLogClientImpl) dlClientBuilder.build();
}
 

void send(SocketAddress address) {
    long elapsedMs = stopwatch.elapsed(TimeUnit.MILLISECONDS);
    if (clientConfig.getMaxRedirects() > 0
        && tries.get() >= clientConfig.getMaxRedirects()) {
        fail(address, new RequestTimeoutException(Duration.fromMilliseconds(elapsedMs),
                "Exhausted max redirects in " + elapsedMs + " ms"));
        return;
    } else if (shouldTimeout(elapsedMs)) {
        fail(address, new RequestTimeoutException(Duration.fromMilliseconds(elapsedMs),
                "Exhausted max request timeout " + clientConfig.getRequestTimeoutMs()
                        + " in " + elapsedMs + " ms"));
        return;
    }
    synchronized (this) {
        String addrStr = address.toString();
        if (ctx.isSetTriedHosts() && ctx.getTriedHosts().contains(addrStr)) {
            nextAddressToSend = address;
            dlTimer.newTimeout(this,
                    Math.min(clientConfig.getRedirectBackoffMaxMs(),
                            tries.get() * clientConfig.getRedirectBackoffStartMs()),
                    TimeUnit.MILLISECONDS);
        } else {
            doSend(address);
        }
    }
}
 

@Test(timeout = 60000)
public void testBuildClientsFromSameBuilder() throws Exception {
    DistributedLogClientBuilder builder = DistributedLogClientBuilder.newBuilder()
            .name("build-clients-from-same-builder")
            .clientId(ClientId$.MODULE$.apply("test-builder"))
            .finagleNameStr("inet!127.0.0.1:7001")
            .streamNameRegex(".*")
            .handshakeWithClientInfo(true)
            .clientBuilder(ClientBuilder.get()
                .hostConnectionLimit(1)
                .connectTimeout(Duration.fromSeconds(1))
                .tcpConnectTimeout(Duration.fromSeconds(1))
                .requestTimeout(Duration.fromSeconds(10)));
    DistributedLogClient client1 = builder.build();
    DistributedLogClient client2 = builder.build();
    assertFalse(client1 == client2);
}
 

@Test(timeout = 60000)
public void testBuildClientsFromSameBuilder() throws Exception {
    DistributedLogClientBuilder builder = DistributedLogClientBuilder.newBuilder()
            .name("build-clients-from-same-builder")
            .clientId(ClientId$.MODULE$.apply("test-builder"))
            .finagleNameStr("inet!127.0.0.1:7001")
            .streamNameRegex(".*")
            .handshakeWithClientInfo(true)
            .clientBuilder(ClientBuilder.get()
                .hostConnectionLimit(1)
                .connectTimeout(Duration.fromSeconds(1))
                .tcpConnectTimeout(Duration.fromSeconds(1))
                .requestTimeout(Duration.fromSeconds(10)));
    DistributedLogClient client1 = builder.build();
    DistributedLogClient client2 = builder.build();
    assertFalse(client1 == client2);
}
 

public void start() {
  Duration[] defaultLatchIntervals = {Duration.apply(1, TimeUnit.MINUTES)};
  @SuppressWarnings("deprecation")
  AdminServiceFactory adminServiceFactory = new AdminServiceFactory(
      this.mPort,
      20,
      List$.MODULE$.<StatsFactory>empty(),
      Option.<String>empty(),
      List$.MODULE$.<Regex>empty(),
      Map$.MODULE$.<String, CustomHttpHandler>empty(),
      List.<Duration>fromArray(defaultLatchIntervals));
  RuntimeEnvironment runtimeEnvironment = new RuntimeEnvironment(this);
  AdminHttpService service = adminServiceFactory.apply(runtimeEnvironment);
  for (Map.Entry<String, CustomHttpHandler> entry : this.mCustomHttpHandlerMap.entrySet()) {
    service.httpServer().createContext(entry.getKey(), entry.getValue());
  }
}
 

public void start() {
  Duration[] defaultLatchIntervals = {Duration.apply(1, TimeUnit.MINUTES)};
  @SuppressWarnings("deprecation")
  AdminServiceFactory adminServiceFactory = new AdminServiceFactory(
      this.mPort,
      20,
      List$.MODULE$.<StatsFactory>empty(),
      Option.<String>empty(),
      List$.MODULE$.<Regex>empty(),
      Map$.MODULE$.<String, CustomHttpHandler>empty(),
      List.<Duration>fromArray(defaultLatchIntervals));
  RuntimeEnvironment runtimeEnvironment = new RuntimeEnvironment(this);
  AdminHttpService service = adminServiceFactory.apply(runtimeEnvironment);
  for (Map.Entry<String, CustomHttpHandler> entry: this.mCustomHttpHandlerMap.entrySet()) {
    service.httpServer().createContext(entry.getKey(), entry.getValue());
  }
}
 

private void startThriftServer(int thriftPort) throws UnknownHostException {
  TerrapinController.ServiceIface serviceImpl = new TerrapinControllerServiceImpl(
      this.configuration,
      this.zkManager,
      this.hdfsClient,
      this.helixAdmin,
      this.clusterName);
  TerrapinController.Service service =
      new TerrapinController.Service(serviceImpl, new TBinaryProtocol.Factory());

  this.server = ServerBuilder.safeBuild(
      service,
      ServerBuilder.get()
      .name("TerrapinController")
      .codec(ThriftServerFramedCodec.get())
      .hostConnectionMaxIdleTime(Duration.fromTimeUnit(
          configuration.getInt(Constants.THRIFT_CONN_MAX_IDLE_TIME, 1), TimeUnit.MINUTES))
      .maxConcurrentRequests(configuration.getInt(Constants.THRIFT_MAX_CONCURRENT_REQUESTS,
          100))
      .reportTo(new OstrichStatsReceiver(Stats.get("")))
      .bindTo(new InetSocketAddress(thriftPort)));
  new OstrichAdminService(configuration.getInt(Constants.OSTRICH_METRICS_PORT, 9999)).start();
}
 

public static void main(String[] args) throws Exception {
  final PropertiesConfiguration config = TerrapinUtil.readPropertiesExitOnFailure(
      System.getProperties().getProperty("terrapin.config", "thrift.properties"));

  OstrichStatsReceiver statsReceiver = new OstrichStatsReceiver(Stats.get(""));
  int listenPort = config.getInt("thrift_port", 9090);
  TerrapinServiceImpl serviceImpl = new TerrapinServiceImpl(config,
      (List) config.getList("cluster_list"));
  Service<byte[], byte[]> service = new TerrapinService.Service(serviceImpl,
      new TBinaryProtocol.Factory());
  Server server = ServerBuilder.safeBuild(
      service,
      ServerBuilder.get()
          .name("TERRAPIN_THRIFT")
          .codec(ThriftServerFramedCodec.get())
          .hostConnectionMaxIdleTime(Duration.apply(1, TimeUnit.MINUTES))
          .maxConcurrentRequests(3000)
          .reportTo(statsReceiver)
          .bindTo(new InetSocketAddress(listenPort)));
  new OstrichAdminService(config.getInt(Constants.OSTRICH_METRICS_PORT, 9999)).start();
  LOG.info("\n#######################################"
          + "\n#      Ready To Serve Requests.       #"
          + "\n#######################################");
}
 

@Test
public void testGetEmptyClusters() {
  ByteBuffer key = ByteBuffer.wrap(KEY);
  TerrapinGetRequest request = prepareGetRequest();
  RequestOptions options = new RequestOptions();
  options.setSelectionPolicy(SelectionPolicy.PRIMARY_FIRST);
  request.setOptions(options);
  request.setClusterList(ImmutableList.copyOf(new String[]{}));

  TerrapinResponse response = new TerrapinResponse().setResponseMap(ImmutableMap.of(
      key, new TerrapinSingleResponse().setValue(ByteBuffer.wrap("value".getBytes()))));
  when(mockClient1.getMany(eq(FILESET), eq(Sets.newHashSet(key)))).thenReturn(
      Future.value(response));
  when(mockClient2.getManyNoRetries(eq(FILESET), eq(Sets.newHashSet(key)))).thenReturn(
      Future.value(response));

  Try<TerrapinSingleResponse> singleResponseTry = serviceIface.get(request).get(Duration.forever());
  assertTrue(singleResponseTry.isThrow());
  assertEquals(TerrapinGetErrorCode.INVALID_REQUEST,
      ((TerrapinGetException)((Throw)singleResponseTry).e()).getErrorCode());

}
 

@Test
public void testMultiGetEmptyClusters() {
  TerrapinMultiGetRequest request = prepareMultiGetRequest();
  RequestOptions options = new RequestOptions();
  options.setSelectionPolicy(SelectionPolicy.PRIMARY_FIRST);
  request.setOptions(options);
  request.setClusterList(ImmutableList.copyOf(new String[]{}));

  TerrapinResponse response = new TerrapinResponse().setResponseMap(ImmutableMap.of(
      ByteBuffer.wrap("key1".getBytes()),
      new TerrapinSingleResponse().setValue("value1".getBytes()),
      ByteBuffer.wrap("key2".getBytes()),
      new TerrapinSingleResponse().setErrorCode(TerrapinGetErrorCode.READ_ERROR)));
  Set<ByteBuffer> keys = Sets.newHashSet(ByteBuffer.wrap("key1".getBytes()),
      ByteBuffer.wrap("key2".getBytes()),
      ByteBuffer.wrap("key3".getBytes()));

  when(mockClient1.getMany(eq(FILESET), eq(keys))).thenReturn(Future.value(response));
  when(mockClient2.getManyNoRetries(eq(FILESET), eq(keys))).thenReturn(Future.value(response));

  Try<TerrapinResponse> returnResponseTry = serviceIface.multiGet(request).get(Duration.forever());
  assertTrue(returnResponseTry.isThrow());
  assertEquals(TerrapinGetErrorCode.INVALID_REQUEST,
      ((TerrapinGetException)((Throw)returnResponseTry).e()).getErrorCode());
}
 

@Test
public void testMultiGetError() {
  TerrapinMultiGetRequest request = prepareMultiGetRequest();
  Set<ByteBuffer> keys = Sets.newHashSet(ByteBuffer.wrap("key1".getBytes()),
      ByteBuffer.wrap("key2".getBytes()),
      ByteBuffer.wrap("key3".getBytes()));

  when (mockClient1.getMany(eq(FILESET), eq(keys))).thenReturn(
      Future.<TerrapinResponse>exception(new TerrapinGetException("Failed",
          TerrapinGetErrorCode.FILE_SET_NOT_FOUND)));
  Try<TerrapinResponse> responseTry = serviceIface.multiGet(request).get(Duration.forever());
  assertTrue(responseTry.isThrow());
  assertEquals(TerrapinGetErrorCode.FILE_SET_NOT_FOUND,
               ((TerrapinGetException)((Throw)responseTry).e()).getErrorCode());

  when(mockClient1.getMany(eq(FILESET), eq(keys))).thenThrow(
      new RuntimeException(new NullPointerException()));
  responseTry = serviceIface.multiGet(request).get(Duration.forever());
  assertTrue(responseTry.isThrow());
  assertEquals(TerrapinGetErrorCode.OTHER,
               ((TerrapinGetException)((Throw)responseTry).e()).getErrorCode());
}
 

private void startThriftServer(int thriftPort) {
  TerrapinServerInternal.ServiceIface serviceImpl = new TerrapinServerInternalImpl(configuration,
      resourcePartitionMap);
  TerrapinServerInternal.Service service =
      new TerrapinServerInternal.Service(serviceImpl, new TBinaryProtocol.Factory());

  this.server = ServerBuilder.safeBuild(
      service,
      ServerBuilder.get()
          .name("TerrapinServer")
          .codec(ThriftServerFramedCodec.get())
          .hostConnectionMaxIdleTime(Duration.fromTimeUnit(
                  configuration.getInt(Constants.THRIFT_CONN_MAX_IDLE_TIME, 1), TimeUnit.MINUTES))
          .maxConcurrentRequests(configuration.getInt(Constants.THRIFT_MAX_CONCURRENT_REQUESTS,
                  100))
          .reportTo(new OstrichStatsReceiver(Stats.get("")))
          .bindTo(new InetSocketAddress(thriftPort)));
    new OstrichAdminService(configuration.getInt(Constants.OSTRICH_METRICS_PORT, 9999)).start();  
}
 

@SuppressWarnings("restriction")
public void start() {
  try {
    Properties properties = new Properties();
    properties.load(this.getClass().getResource("build.properties").openStream());
    LOG.info("build.properties build_revision: {}",
        properties.getProperty("build_revision", "unknown"));
  } catch (Throwable t) {
    LOG.warn("Failed to load properties from build.properties");
  }
  Duration[] defaultLatchIntervals = {Duration.apply(1, TimeUnit.MINUTES)};
  Iterator<Duration> durationIterator = Arrays.asList(defaultLatchIntervals).iterator();

  AdminServiceFactory adminServiceFactory = new AdminServiceFactory(
          this.port,
          20,
          List$.MODULE$.<StatsFactory>empty(),
          Option.<String>empty(),
          List$.MODULE$.<Regex>empty(),
          Map$.MODULE$.<String, CustomHttpHandler>empty(),
          JavaConversions.asScalaIterator(durationIterator).toList());

  RuntimeEnvironment runtimeEnvironment = new RuntimeEnvironment(this);
  AdminHttpService service = adminServiceFactory.apply(runtimeEnvironment);
  for (Map.Entry<String, CustomHttpHandler> entry : this.customHttpHandlerMap.entrySet()) {
    service.httpServer().createContext(entry.getKey(), entry.getValue());
  }
}
 

public void openLogWriter() throws Throwable {
    try {
        closeLogWriter();
        logWriter = FutureUtils.result(logManager.openAsyncLogWriter(), Duration.fromSeconds(DL_OPERATION_TIMEOUT_SECS));
    } catch (Throwable t) {
        logger.error("errors while opening log writer", t);
        throw t;
    }
}
 

public void openLogReader(DLSN dlsn) throws IOException {
    try {
        logReader = FutureUtils.result(logManager.openAsyncLogReader(dlsn), Duration.fromSeconds(DL_OPERATION_TIMEOUT_SECS));
    } catch (IOException e) {
        logger.error("errors while open or create log service", e);
        throw e;
    }
}
 

public void shutdown() {
  this.server.close(Duration.fromSeconds(10));
  this.helixManager.disconnect();
  // Sleep a little to make sure that the zk node truly disappears.
  try {
    Thread.sleep(5000);
  } catch (InterruptedException e) {
    LOG.error("Interrupted in shutdown. This should not happen.");
  }
  this.resourcePartitionMap.close();
}
 

@RequestMapping(value = "/search", method = RequestMethod.GET)
@ResponseBody
public String search(@RequestParam("bql") String bql) {
  LindenResult result;
  try {
    Future<LindenResult> future = LindenAdmin.getService().handleClusterSearchRequest(bql);
    result = Await.result(future, Duration.apply(30000, TimeUnit.MILLISECONDS));
  } catch (Exception e) {
    result = new LindenResult();
    result.setSuccess(false).setError(Throwables.getStackTraceAsString(e));
  }
  return ThriftToJSON(result);
}
 

@RequestMapping(value = "/index", method = RequestMethod.POST)
@ResponseBody
public String index(@RequestParam("content") String content) {
  Response response;
  try {
    Future<Response> future = LindenAdmin.getService().handleClusterIndexRequest(content);
    response = Await.result(future, Duration.apply(30000, TimeUnit.MILLISECONDS));
  } catch (Exception e) {
    response = new Response();
    response.setSuccess(false).setError(Throwables.getStackTraceAsString(e));
  }
  return ThriftToJSON(response);
}
 

@RequestMapping(value = "/delete", method = RequestMethod.POST)
@ResponseBody
public String delete(@RequestParam("bql") String bql) {
  Response response;
  try {
    Future<Response> future = LindenAdmin.getService().handleClusterDeleteRequest(bql);
    response = Await.result(future, Duration.apply(30000, TimeUnit.MILLISECONDS));
  } catch (Exception e) {
    response = new Response();
    response.setSuccess(false).setError(Throwables.getStackTraceAsString(e));
  }
  return ThriftToJSON(response);
}
 

public static void main(String[] args) {
  try {
    String serverHostName = InetAddress.getLocalHost().getHostName();
    PinLaterQueueConfig queueConfig = new PinLaterQueueConfig(CONFIGURATION);
    queueConfig.initialize();
    String backend = CONFIGURATION.getString("PINLATER_BACKEND");
    PinLaterBackendIface backendIFace = getBackendIface(backend, serverHostName);
    PinLaterServiceImpl serviceImpl = new PinLaterServiceImpl(backendIFace, queueConfig);
    PinLater.Service service = new PinLater.Service(serviceImpl, new TBinaryProtocol.Factory());
    ServiceShutdownHook.register(ServerBuilder.safeBuild(
        service,
        ServerBuilder.get()
            .name("PinLaterService")
            .codec(ThriftServerFramedCodec.get())
            .hostConnectionMaxIdleTime(Duration.fromTimeUnit(
                CONFIGURATION.getInt("SERVER_CONN_MAX_IDLE_TIME_MINUTES"), TimeUnit.MINUTES))
            .maxConcurrentRequests(CONFIGURATION.getInt("MAX_CONCURRENT_REQUESTS"))
            .reportTo(new OstrichStatsReceiver(Stats.get("")))
            .bindTo(new InetSocketAddress(CONFIGURATION.getInt("THRIFT_PORT")))));
    new OstrichAdminService(CONFIGURATION.getInt("OSTRICH_PORT")).start();

    LOG.info("\n#######################################"
        + "\n#      Ready To Serve Requests.       #"
        + "\n#######################################");
  } catch (Exception e) {
    LOG.error("Failed to start the pinlater server", e);
    System.exit(1);
  }
}
 

private Future<Integer> getFuture(final long futureExecutionTimeMs,
                                 final Integer futureValue,
                                 final boolean isFutureSuccessful) {
 return timer.doLater(
     Duration.fromMilliseconds(futureExecutionTimeMs),
       new Function0<Integer>() {
         public Integer apply() {
           if (isFutureSuccessful) {
             return futureValue;
           } else {
             throw new RuntimeException(EXCEPTION_MSG);
           }
         }
       });
}
 

static DistributedLogClientBuilder createDistributedLogClientBuilder(ServerSet serverSet) {
    return DistributedLogClientBuilder.newBuilder()
                    .name("rebalancer_tool")
                    .clientId(ClientId$.MODULE$.apply("rebalancer_tool"))
                    .maxRedirects(2)
                    .serverSet(serverSet)
                    .clientBuilder(ClientBuilder.get()
                            .connectionTimeout(Duration.fromSeconds(2))
                            .tcpConnectTimeout(Duration.fromSeconds(2))
                            .requestTimeout(Duration.fromSeconds(10))
                            .hostConnectionLimit(1)
                            .hostConnectionCoresize(1)
                            .keepAlive(true)
                            .failFast(false));
}
 

public PlacementPolicy(LoadAppraiser loadAppraiser, RoutingService routingService,
                       Namespace namespace, PlacementStateManager placementStateManager,
                       Duration refreshInterval, StatsLogger statsLogger) {
    this.loadAppraiser = loadAppraiser;
    this.routingService = routingService;
    this.namespace = namespace;
    this.placementStateManager = placementStateManager;
    this.refreshInterval = refreshInterval;
    placementCalcStats = statsLogger.getOpStatsLogger("placement");
}