import com.twitter.util.Duration; //导入依赖的package包/类
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());
  }
}
 

import com.twitter.util.Duration; //导入依赖的package包/类
/**
 * Constructs a finagle java memcached client for the list of endpoints..
 *
 * @param endpoints list of {@code InetSocketAddress} for all the memcached servers.
 * @return {@link Client} to read/write to the hash ring of the servers..
 */
static Client makeMemcachedClient(Options opts, List<InetSocketAddress> endpoints)
    throws UnknownHostException {
  com.twitter.finagle.memcached.Client client =
      KetamaClientBuilder.get()
          .nodes(getHostPortWeightTuples(endpoints))
          .clientBuilder(ClientBuilder.get()
                             .codec(new Memcached())
                             .tcpConnectTimeout(new Duration(TimeUnit.MILLISECONDS.toNanos(opts.connectTimeoutMillis)))
                             .requestTimeout(new Duration(TimeUnit.MILLISECONDS.toNanos(opts.requestTimeoutMillis)))
                             .timeout(new Duration(TimeUnit.MILLISECONDS.toNanos(opts.e2eTimeoutMillis)))
                             .hostConnectionLimit(opts.hostConnectionLimit)
                             .hostConnectionMaxWaiters(opts.maxWaiters)
                             .retries(opts.requestRetries))
          .build();

  return new ClientBase(client);
}
 

import com.twitter.util.Duration; //导入依赖的package包/类
@Test public void flushesIncompleteSpans() throws Exception {
  advanceAndRecord(0, root, new Annotation.Rpc("GET"));
  advanceAndRecord(15, root, new Annotation.ServiceName("frontend"));
  advanceAndRecord(0, root, new Annotation.ServerRecv());
  // Note: there's no ServerSend() which would complete the span.

  time.advance(recorder.ttl.plus(Duration.fromMilliseconds(1))); // advance timer
  timer.tick(); // invokes a flush

  Span span = spansSent.take();
  assertThat(span.idString()).isEqualTo(root.toString());
  assertThat(span.name).isEqualTo("get");
  assertThat(span.annotations).extracting(a -> a.value).containsExactly(
      "sr", "finagle.flush"
  );
  assertThat(span.duration).isNull();
}
 

import com.twitter.util.Duration; //导入依赖的package包/类
@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);
}
 

import com.twitter.util.Duration; //导入依赖的package包/类
/**
 * Sanity check to make sure both checksum flag values work.
 */
@Test(timeout = 60000)
public void testChecksumFlag() throws Exception {
    String name = "dlserver-basic-write";
    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 = (DistributedLogClientImpl) 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();
}
 

import com.twitter.util.Duration; //导入依赖的package包/类
@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));
    }
}
 

import com.twitter.util.Duration; //导入依赖的package包/类
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();
}
 

import com.twitter.util.Duration; //导入依赖的package包/类
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());
  }
}
 

import com.twitter.util.Duration; //导入依赖的package包/类
/**
 * Call this when each request completes.
 *
 * @param latency Latency for that request.
 */
public synchronized void requestComplete(Duration latency) {
  queriesIssued++;
  totalLatencyMicros += latency.inMicroseconds();

  long currentTimeMillis = System.currentTimeMillis();
  if (currentTimeMillis >= lastLogTimestampMillis + logIntervalSeconds * 1000) {
    long qps = queriesIssued * 1000 / (currentTimeMillis - lastLogTimestampMillis);
    double avgLatencyMillis = totalLatencyMicros / (queriesIssued * 1000.0);
    LOG.info(String.format("QPS: %d Avg Latency (ms): %.5f", qps, avgLatencyMillis));

    queriesIssued = 0;
    totalLatencyMicros = 0;
    lastLogTimestampMillis = currentTimeMillis;
  }
}
 

import com.twitter.util.Duration; //导入依赖的package包/类
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();
}
 

import com.twitter.util.Duration; //导入依赖的package包/类
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#######################################");
}
 

import com.twitter.util.Duration; //导入依赖的package包/类
@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());

}
 

import com.twitter.util.Duration; //导入依赖的package包/类
@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());
}
 

import com.twitter.util.Duration; //导入依赖的package包/类
@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());
}
 

import com.twitter.util.Duration; //导入依赖的package包/类
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();  
}
 

import com.twitter.util.Duration; //导入依赖的package包/类
@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());
}
 

import com.twitter.util.Duration; //导入依赖的package包/类
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());
  }
}
 

import com.twitter.util.Duration; //导入依赖的package包/类
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);
    adminServiceFactory.apply(runtimeEnvironment);
    try {
        Properties properties = new Properties();
        properties.load(this.getClass().getResource("build.properties").openStream());
        String buildRevision = properties.getProperty("build_revision", "unknown");
        LOG.info("build.properties build_revision: {}",
                 properties.getProperty("build_revision", "unknown"));
        StatsUtil.setLabel("secor.build_revision", buildRevision);
    } catch (Throwable t) {
        LOG.error("Failed to load properties from build.properties", t);
    }
}
 

import com.twitter.util.Duration; //导入依赖的package包/类
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;
    }
}
 

import com.twitter.util.Duration; //导入依赖的package包/类
public DLSN getFirstLogDLSN() throws Throwable {
    try {
        return FutureUtils.result(logManager.getFirstDLSNAsync(), Duration.fromSeconds(DL_OPERATION_TIMEOUT_SECS));
    } catch (Throwable t) {
        if (t instanceof LogEmptyException) {
            logger.info("log is empty, just return initial dlsn");
            return DLSN.InitialDLSN;
        } else {
            throw t;
        }
    }
}