Java 类org.apache.thrift.transport.TTransportException 实例源码

private static TServerSocket createServerTransport(boolean secure)
        throws TTransportException
{
    if (!secure) {
        return new TServerSocket(0);
    }

    try {
        SSLContext serverSslContext = ClientTestUtils.getServerSslContext();
        SSLServerSocket serverSocket = (SSLServerSocket) serverSslContext.getServerSocketFactory().createServerSocket(0);
        return new TServerSocket(serverSocket);
    }
    catch (Exception e) {
        throw new TTransportException("Error initializing secure socket", e);
    }
}

private static TSocket createClientSocket(boolean secure, HostAndPort address)
        throws TTransportException
{
    if (!secure) {
        return new TSocket(address.getHost(), address.getPort());
    }

    try {
        SSLContext serverSslContext = ClientTestUtils.getClientSslContext();
        SSLSocket clientSocket = (SSLSocket) serverSslContext.getSocketFactory().createSocket(address.getHost(), address.getPort());
        //            clientSocket.setSoTimeout(timeout);
        return new TSocket(clientSocket);
    }
    catch (Exception e) {
        throw new TTransportException("Error initializing secure socket", e);
    }
}

@SuppressWarnings("deprecation")
@Override
public void start() {
  try {
    InetSocketAddress bindAddr = new InetSocketAddress(host, port);
    serverTransport = new TServerSocket(bindAddr);
    ThriftFlumeEventServer.Processor processor =
        new ThriftFlumeEventServer.Processor(new ThriftFlumeEventServerImpl());
    server = new TThreadPoolServer(
        new TThreadPoolServer.Args(serverTransport).processor(processor));
  } catch (TTransportException e) {
    throw new FlumeException("Failed starting source", e);
  }
  ThriftHandler thriftHandler = new ThriftHandler(server);
  thriftHandlerThread = new Thread(thriftHandler);
  thriftHandlerThread.start();
  super.start();
}

@Override
public Pair<TTransport, Bmv2DeviceThriftClient> load(DeviceId deviceId)
        throws TTransportException {
    log.debug("Instantiating new client... > deviceId={}", deviceId);
    // Make the expensive call
    Bmv2Device device = Bmv2Device.of(deviceId);
    TTransport transport = new TSocket(device.thriftServerHost(), device.thriftServerPort());
    TProtocol protocol = new TBinaryProtocol(transport);
    // Our BMv2 device implements multiple Thrift services, create a client for each one on the same transport.
    Standard.Client standardClient = new Standard.Client(
            new TMultiplexedProtocol(protocol, "standard"));
    SimpleSwitch.Client simpleSwitch = new SimpleSwitch.Client(
            new TMultiplexedProtocol(protocol, "simple_switch"));
    // Wrap clients so to automatically have synchronization and resiliency to connectivity errors
    Standard.Iface safeStandardClient = SafeThriftClient.wrap(standardClient,
                                                              Standard.Iface.class,
                                                              options);
    SimpleSwitch.Iface safeSimpleSwitchClient = SafeThriftClient.wrap(simpleSwitch,
                                                                      SimpleSwitch.Iface.class,
                                                                      options);
    Bmv2DeviceThriftClient client = new Bmv2DeviceThriftClient(deviceId,
                                                               transport,
                                                               safeStandardClient,
                                                               safeSimpleSwitchClient);
    return Pair.of(transport, client);
}

@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

    // Thrift transport layer is not thread-safe (it's a wrapper on a socket), hence we need locking.
    synchronized (transport) {

        LOG.debug("Invoking method... > fromThread={}, method={}, args={}",
                  Thread.currentThread().getId(), method.getName(), args);

        try {

            return method.invoke(baseClient, args);
        } catch (InvocationTargetException e) {
            if (e.getTargetException() instanceof TTransportException) {
                TTransportException cause = (TTransportException) e.getTargetException();

                if (RESTARTABLE_CAUSES.contains(cause.getType())) {
                    // Try to reconnect. If fail, a TTransportException will be thrown.
                    reconnectOrThrowException();
                    try {
                        // If here, transport has been successfully open, hence new exceptions will be thrown.
                        return method.invoke(baseClient, args);
                    } catch (InvocationTargetException e1) {
                        LOG.debug("Exception: {}", e1.getTargetException());
                        throw e1.getTargetException();
                    }
                }
            }
            // Target exception is neither a TTransportException nor a restartable cause.
            LOG.debug("Exception: {}", e.getTargetException());
            throw e.getTargetException();
        }
    }
}

@Override
public void flush() throws TTransportException {
    synchronized (sendLock) {
        byte[] bytes = new byte[MAX_BUFFER_SIZE];
        int length = writeBuffer.position();

        writeBuffer.flip();
        writeBuffer.get(bytes, 0, length);

        try {
            socket.send(new DatagramPacket(bytes, length));
        } catch (IOException e) {
            throw new TTransportException(e);
        } finally {
            writeBuffer.clear();
        }
    }
}

@Override
public void write(byte[] bytes, int offset, int length) throws TTransportException {
    if (!isOpen()) {
        throw new TTransportException(TTransportException.NOT_OPEN);
    }

    synchronized (sendLock) {
        if (writeBuffer.position() + length > MAX_BUFFER_SIZE) {
            throw new TTransportException(
                String.format("Message size too large: %d is greater than available size %d",
                    length,
                    MAX_BUFFER_SIZE - writeBuffer.position()
                )
            );
        }

        writeBuffer.put(bytes, offset, length);
    }
}

public void emitMetricBatch(MetricBatch batch) throws TTransportException {
    lock.writeLock().lock();

    batches.add(batch);

    if (countBatches) {
        phaser.arrive();
    }

    for (Metric metric : batch.getMetrics()) {
        metrics.add(metric);

        if (!countBatches) {
            phaser.arrive();
        }
    }

    lock.writeLock().unlock();

    throw new TTransportException(TTransportException.END_OF_FILE, "complete");
}

private TServerSocket createServerSocket(boolean useSSL, int port) throws IOException, TTransportException {
  TServerSocket serverSocket = null;
  // enable SSL support for HMS
  List<String> sslVersionBlacklist = new ArrayList<>();
  for (String sslVersion : hiveConf.getVar(ConfVars.HIVE_SSL_PROTOCOL_BLACKLIST).split(",")) {
    sslVersionBlacklist.add(sslVersion);
  }
  if (!useSSL) {
    serverSocket = HiveAuthUtils.getServerSocket(null, port);
  } else {
    String keyStorePath = hiveConf.getVar(ConfVars.HIVE_METASTORE_SSL_KEYSTORE_PATH).trim();
    if (keyStorePath.isEmpty()) {
      throw new IllegalArgumentException(
          ConfVars.HIVE_METASTORE_SSL_KEYSTORE_PASSWORD.varname + " Not configured for SSL connection");
    }
    String keyStorePassword = ShimLoader.getHadoopShims().getPassword(hiveConf,
        HiveConf.ConfVars.HIVE_METASTORE_SSL_KEYSTORE_PASSWORD.varname);
    serverSocket = HiveAuthUtils.getServerSSLSocket(null, port, keyStorePath, keyStorePassword, sslVersionBlacklist);
  }
  return serverSocket;
}

ServerThread() throws TTransportException {
   TMultiplexedProcessor processor = new TMultiplexedProcessor();
   for (String beanName : serviceMap.keySet()) {
      IThriftServerService serverService = (IThriftServerService) serviceMap.getService(beanName);
      String processorName = serverService.getName();
      TProcessor tProcessor = serverService.getProcessor(serverService);
      processor.registerProcessor(processorName, tProcessor);
      logger.info("Register a processorName {} processorImpl {}", processorName, tProcessor);
   }

   logger.info("init default TServerTransport in addr {} port {}", applicationProperties.getAddr(), applicationProperties.getPort());
   TServerTransport tServerTransport = new TServerSocket(new InetSocketAddress(applicationProperties.getAddr(),
           applicationProperties.getPort()));
   TThreadPoolServer.Args args = new TThreadPoolServer.Args(tServerTransport);
   args.processor(processor);
   args.protocolFactory(tProtocolFactory);
   server = new TThreadPoolServer(args);
}

public TsfileConnection(String url, Properties info) throws SQLException, TTransportException {
if (url == null) {
    throw new TsfileURLException("Input url cannot be null");
}
params = Utils.parseURL(url, info);

supportedProtocols.add(TSProtocolVersion.TSFILE_SERVICE_PROTOCOL_V1);

openTransport();
client = new TSIService.Client(new TBinaryProtocol(transport));
// open client session
openSession();
// Wrap the client with a thread-safe proxy to serialize the RPC calls
client = newSynchronizedClient(client);
autoCommit = false;
   }

private static TServer getTThreadPoolServer(TProtocolFactory protocolFactory,
                                            TProcessor processor,
                                            TTransportFactory transportFactory,
                                            int workerThreads,
                                            InetSocketAddress inetSocketAddress,
                                            int backlog,
                                            int clientTimeout)
    throws TTransportException {
  TServerTransport serverTransport = new TServerSocket(
                                         new TServerSocket.ServerSocketTransportArgs().
                                             bindAddr(inetSocketAddress).backlog(backlog).
                                             clientTimeout(clientTimeout));
  log.info("starting HBase ThreadPool Thrift server on " + inetSocketAddress.toString());
  TThreadPoolServer.Args serverArgs = new TThreadPoolServer.Args(serverTransport);
  serverArgs.processor(processor);
  serverArgs.transportFactory(transportFactory);
  serverArgs.protocolFactory(protocolFactory);
  if (workerThreads > 0) {
    serverArgs.maxWorkerThreads(workerThreads);
  }
  return new TThreadPoolServer(serverArgs);
}

/**
 * Ensures that the Mock Cassandra instance is up and running. Will reinit
 * the database every time it is called.
 *
 * @param cassandraKeyspace Cassandra keyspace to setup.
 * @return A cluster object.
 * @throws ConfigurationException
 * @throws IOException
 * @throws InterruptedException
 * @throws TTransportException
 */
public static Cluster ensureMockCassandraRunningAndEstablished(String cassandraKeyspace) throws ConfigurationException, IOException, InterruptedException, TTransportException
{
    Cluster cluster;
    long timeout = 60000;
    EmbeddedCassandraServerHelper.startEmbeddedCassandra(timeout);
    cluster = Cluster.builder().addContactPoints("127.0.0.1").withPort(9142).build();
    //Thread.sleep(20000);//time to let cassandra startup
    final Metadata metadata = cluster.getMetadata();

    Session session = cluster.connect();
    Utils.initDatabase(DB_CQL, session);
    session = cluster.connect(cassandraKeyspace);

    logger.info("Connected to cluster: " + metadata.getClusterName() + '\n');
    return cluster;
}

public InetAddress resolveHost() throws TTransportException {
  // TODO: SW: JDK's InetAddress has an inbuilt cache but it is no good
  // and does not honour DNS TTL etc. Custom DNSCacheService should be
  // added for better behaviour.

  // InetAddress addr = DNSCacheService.getInstance().lookupCache(hostName);
  try {
    return InetAddress.getByName(this.hostName);
  } catch (UnknownHostException uhe) {
    // use ipAddress if available
    if (this.ipAddress != null) {
      try {
        return InetAddress.getByName(this.ipAddress);
      } catch (UnknownHostException e) {
        throw new TTransportException(e);
      }
    }
    else {
      throw new TTransportException(uhe);
    }
  }
}

@Override
public Client create() throws Exception {
  TSocket transport = new TSocket(host, port);
  try {
    transport.open();
  } catch (TTransportException e) {
    throw new InterpreterException(e);
  }

  TProtocol protocol = new  TBinaryProtocol(transport);
  Client client = new RemoteInterpreterService.Client(protocol);

  synchronized (clientSocketMap) {
    clientSocketMap.put(client, transport);
  }
  return client;
}

/**
 * Sets the socket properties like timeout, keepalive, buffer sizes.
 * 
 * @param timeout
 *          Milliseconds timeout
 * @param params
 *          Socket parameters including buffer sizes and keep-alive settings
 * @param props
 *          the system properties instance to use and initialize global socket
 *          options like keepalive and buffer sizes that are not set in params
 */
protected void setProperties(Socket socket, int timeout,
    SocketParameters params, SystemProperties props)
    throws TTransportException {
  this.inputBufferSize = params.getInputBufferSize(props
      .getSocketInputBufferSize());
  this.outputBufferSize = params.getOutputBufferSize(props
      .getSocketOutputBufferSize());
  try {
    socket.setSoLinger(false, 0);
    socket.setTcpNoDelay(true);
    this.timeout = GfxdTSocket.setTimeout(socket, timeout, params, props);
  } catch (SocketException se) {
    LOGGER.warn("Could not set socket timeout.", se);
    throw new TTransportException(TTransportException.NOT_OPEN,
        "Could not set socket timeout.", se);
  }
}

/**
 * Initializes the socket object
 */
private static SocketChannel initSocket(boolean blocking)
    throws TTransportException {
  try {
    SocketChannel socketChannel = SocketChannel.open();
    socketChannel.configureBlocking(blocking);
    return socketChannel;
  } catch (SocketException se) {
    LOGGER.error("Could not configure socket.", se);
    throw new TTransportException(TTransportException.NOT_OPEN,
        "Could not configure socket.", se);
  } catch (IOException ioe) {
    throw new TTransportException(TTransportException.NOT_OPEN,
        "Could not open socket channel.", ioe);
  }
}

/**
 * Get a configured SSL socket connected to the specified host and port.
 * <p>
 * If SSLSocketParameters are not null, then they are used to set the values
 * for the algorithms, keystore, truststore and other settings.
 * <p>
 * Else if SSLSocketParameters are null then the default settings are used.
 * Default settings are retrieved from System properties that are set.
 * 
 * Example system properties: -Djavax.net.ssl.trustStore=<truststore location>
 * -Djavax.net.ssl.trustStorePassword=password
 * -Djavax.net.ssl.keyStore=<keystore location>
 * -Djavax.net.ssl.keyStorePassword=password
 * <p>
 * All the client methods return a bound connection, so there is no need to
 * call open() on the TTransport.
 */
public static SSLSocket getClientSocket(InetAddress hostAddress, int port,
    int timeout, SocketParameters params) throws TTransportException {
  if (params == null) {
    SSLSocketFactory factory = (SSLSocketFactory)SSLSocketFactory
        .getDefault();
    return createClient(factory, hostAddress, port, timeout, null);
  }
  else {
    if (!(params.isSSLKeyStoreSet() || params.isSSLTrustStoreSet())) {
      throw new TTransportException(
          "Either one of the KeyStore or TrustStore must be set "
              + "for SSLSocketParameters");
    }

    SSLContext ctx = createSSLContext(params);
    return createClient(ctx.getSocketFactory(), hostAddress, port, timeout,
        params);
  }
}

private static SSLSocket createClient(SSLSocketFactory factory,
    InetAddress hostAddress, int port, int timeout,
    final SocketParameters params) throws TTransportException {
  try {
    SSLSocket socket = (SSLSocket)factory.createSocket(hostAddress, port);
    socket.setSoTimeout(timeout);
    if (params != null) {
      if (params.getSSLEnabledProtocols() != null) {
        socket.setEnabledProtocols(params.getSSLEnabledProtocols());
      }
      if (params.getSSLCipherSuites() != null) {
        socket.setEnabledCipherSuites(params.getSSLCipherSuites());
      }
    }
    return socket;
  } catch (IOException ioe) {
    throw new TTransportException(TTransportException.NOT_OPEN, ioe);
  } catch (Exception e) {
    throw new TTransportException(TTransportException.NOT_OPEN,
        "Could not connect to " + hostAddress + " on port " + port, e);
  }
}

/**
 * Creates a server socket from underlying socket object
 */
public GfxdTSSLServerSocket(ServerSocket serverSocket,
    InetSocketAddress bindAddress, SocketParameters params)
    throws TTransportException {
  this.socketParams = params;
  try {
    this.serverSocket = serverSocket;
    // Prevent 2MSL delay problem on server restarts
    serverSocket.setReuseAddress(true);
    // Bind to listening port
    if (!serverSocket.isBound()) {
      // backlog hardcoded to 100 as in TSSLTransportFactory
      serverSocket.bind(bindAddress, 100);
    }
  } catch (IOException ioe) {
    throw new TTransportException(TTransportException.NOT_OPEN,
        "Could not bind to host:port " + bindAddress.toString(), ioe);
  }
}

/**
 * Get a configured SSL wrapped TServerSocket bound to the specified port and
 * interface.
 * <p>
 * If SocketParameters have SSL properties set, then they are used to set the
 * values for the algorithms, keystore, truststore and other settings.
 * <p>
 * Else if SocketParameters don't have SSL settings, then the default settings
 * are used. Default settings are retrieved from server System properties.
 * 
 * Example system properties: -Djavax.net.ssl.trustStore=<truststore location>
 * -Djavax.net.ssl.trustStorePassword=password
 * -Djavax.net.ssl.keyStore=<keystore location>
 * -Djavax.net.ssl.keyStorePassword=password
 * 
 * 
 * @return An SSL wrapped {@link GfxdTSSLServerSocket}
 */
public static GfxdTSSLServerSocket getServerSocket(
    InetSocketAddress bindAddress, SocketParameters params)
    throws TTransportException {
  if (params.hasSSLParams()) {
    if (!params.isSSLKeyStoreSet() && !params.isSSLTrustStoreSet()) {
      throw new TTransportException(
          "Either one of the KeyStore or TrustStore must be set "
              + "for SocketParameters having SSL parameters");
    }

    SSLContext ctx = GfxdTSSLSocketFactory.createSSLContext(params);
    return createServer(ctx.getServerSocketFactory(), bindAddress, params);
  }
  else {
    SSLServerSocketFactory factory = (SSLServerSocketFactory)SSLServerSocketFactory
        .getDefault();
    return createServer(factory, bindAddress, params);
  }
}

private static GfxdTSSLServerSocket createServer(
    SSLServerSocketFactory factory, InetSocketAddress bindAddress,
    SocketParameters params) throws TTransportException {
  try {
    SSLServerSocket serverSocket = (SSLServerSocket)factory
        .createServerSocket(bindAddress.getPort(), 100,
            bindAddress.getAddress());
    if (params != null) {
      if (params.getSSLEnabledProtocols() != null) {
        serverSocket.setEnabledProtocols(params.getSSLEnabledProtocols());
      }
      if (params.getSSLCipherSuites() != null) {
        serverSocket.setEnabledCipherSuites(params.getSSLCipherSuites());
      }
      serverSocket.setNeedClientAuth(params.getSSLClientAuth());
    }
    return new GfxdTSSLServerSocket(serverSocket, bindAddress, params);
  } catch (Exception e) {
    throw new TTransportException(TTransportException.NOT_OPEN,
        "Could not bind to host:port " + bindAddress.toString(), e);
  }
}

/**
 * Creates a port listening server socket
 */
public GfxdTServerSocket(InetSocketAddress bindAddress, boolean blocking,
    boolean clientBlocking, SocketParameters params)
    throws TTransportException {
  this.clientBlocking = clientBlocking;
  this.socketParams = params;
  try {
    // Make server socket
    this.serverSockChannel = ServerSocketChannel.open();
    this.serverSockChannel.configureBlocking(blocking);
    ServerSocket socket = this.serverSockChannel.socket();
    // Prevent 2MSL delay problem on server restarts
    socket.setReuseAddress(true);
    // Bind to listening port
    socket.bind(bindAddress);
  } catch (IOException ioe) {
    throw new TTransportException(TTransportException.NOT_OPEN,
        "Could not bind to host:port " + bindAddress.toString(), ioe);
  }
}

@BeforeClass
public static void startServer() throws InterruptedException, TTransportException, ConfigurationException, IOException {
    EmbeddedCassandraServerHelper.startEmbeddedCassandra();
    Cluster cluster = new Cluster.Builder().addContactPoints("127.0.0.1").withPort(9142).build();
    Session session = cluster.connect();
    CQLDataLoader dataLoader = new CQLDataLoader(session);
    dataLoader.load(new ClassPathCQLDataSet("config/cql/create-tables.cql", true, "cassandra_unit_keyspace"));
}

public static void main(String argvs[]) throws TTransportException, IOException {
    Asserts.check(argvs != null && argvs.length == 2, "require params type and config path");
    String type = argvs[0];
    String config = argvs[1];

    try {
        if ("server".equals(type)) {
            ServerForSync.main(new String[] { config });
        } else if ("client".equals(type)) {
            ClientForSync.main(new String[] { config });
        } else if ("client_sync".equals(type)) {
            ClientForSync.sync(config);
        } else if ("client_validate".equals(type)) {
            ClientForSync.validate(config);
        } else {
            throw new RuntimeException("unknow type " + type);
        }
    } finally {
        ThriftClientPool.closeAll();
    }
}

public void start() throws TTransportException {

        synchronized (this) {
            if (null != tserver) {
                throw new RuntimeException("Server is start!");
            }

            SyncFileServerHandler handler = new SyncFileServerHandler(this);
            SyncFileServer.Processor<SyncFileServerHandler> processor = new SyncFileServer.Processor<SyncFileServerHandler>(
                    handler);

            if (null == this.keystore) {
                tserver = simple(processor, port, type);
            } else {
                tserver = secure(processor, port, type, keystore);
            }
        }

        tserver.serve();
    }

public Worker(int id, long durationMillis, String path, int globals) throws IOException {
    this.id = id;
    this.workerDuration = durationMillis;
    this.localPath = path + "/f" + Integer.toString(id);
    this.globalPath = "/f" + Integer.toString(id);
    this.instanceMap = new HashMap<>();
    this.globals = globals;

    String replicaHost = replicaAddr.split(":")[0];
    int replicaPort = Integer.parseInt(replicaAddr.split(":")[1]);
    TTransport transport = new TSocket(replicaHost, replicaPort);
    try {
        transport.open();
    } catch (TTransportException e) {
        throw new RuntimeException(e);
    }
    TProtocol protocol = new TBinaryProtocol(transport);
    c = new FuseOps.Client(protocol);
    out = new BufferedWriter(new FileWriter(new File(logPrefix + this.id)));
}

public Worker(int id, long durationMillis, String path, int globals) throws IOException {
    this.id = id;
    this.workerDuration = durationMillis;
    this.path = path;
    this.globals = globals;

    String replicaHost = replicaAddr.split(":")[0];
    int replicaPort = Integer.parseInt(replicaAddr.split(":")[1]);
    TTransport transport = new TSocket(replicaHost, replicaPort);
    try {
        transport.open();
    } catch (TTransportException e) {
        throw new RuntimeException(e);
    }
    TProtocol protocol = new TBinaryProtocol(transport);
    c = new FuseOps.Client(protocol);
    out = new BufferedWriter(new FileWriter(new File(logPrefix + this.id)));
}

public Worker(int id, long durationMillis, String path, int globals) throws IOException {
    this.id = id;
    this.workerDuration = durationMillis;
    this.localPath = path + "/f" + Integer.toString(id);
    this.globalPath = "/f" + Integer.toString(id);
    this.instanceMap = new HashMap<>();
    this.globals = globals;

    String replicaHost = replicaAddr.split(":")[0];
    int replicaPort = Integer.parseInt(replicaAddr.split(":")[1]);
    TTransport transport = new TSocket(replicaHost, replicaPort);
    try {
        transport.open();
    } catch (TTransportException e) {
        throw new RuntimeException(e);
    }
    TProtocol protocol = new TBinaryProtocol(transport);
    c = new FuseOps.Client(protocol);
    out = new BufferedWriter(new FileWriter(new File(logPrefix + this.id)));
}

public Worker(int id, long durationMillis, String path, int globals) throws IOException {
    this.id = id;
    this.workerDuration = durationMillis;
    this.localPath = path + "/f" + Integer.toString(id);
    this.globalPath = "/f" + Integer.toString(id);
    this.instanceMap = new HashMap<>();
    this.globals = globals;

    String replicaHost = replicaAddr.split(":")[0];
    int replicaPort = Integer.parseInt(replicaAddr.split(":")[1]);
    TTransport transport = new TSocket(replicaHost, replicaPort);
    try {
        transport.open();
    } catch (TTransportException e) {
        throw new RuntimeException(e);
    }
    TProtocol protocol = new TBinaryProtocol(transport);
    c = new FuseOps.Client(protocol);
    out = new BufferedWriter(new FileWriter(new File(logPrefix + this.id)));
}

public Worker(int id, String path) throws IOException {
    this.id = id;
    this.path = path;
    this.localPath = path + "/f" + Integer.toString(id);
    this.globalPath = "/f" + Integer.toString(id);
    this.instanceMap = new HashMap<>();

    String replicaHost = replicaAddr.split(":")[0];
    int replicaPort = Integer.parseInt(replicaAddr.split(":")[1]);
    TTransport transport = new TSocket(replicaHost, replicaPort);
    try {
        transport.open();
    } catch (TTransportException e) {
        throw new RuntimeException(e);
    }
    TProtocol protocol = new TBinaryProtocol(transport);
    c = new FuseOps.Client(protocol);
}

public Worker(int id, long durationMillis, String path, int globals) throws IOException {
    this.id = id;
    this.workerDuration = durationMillis;
    this.localPath = path + "/f" + Integer.toString(id);
    this.globalPath = "/f" + Integer.toString(id);
    this.instanceMap = new HashMap<>();
    this.globals = globals;

    String replicaHost = replicaAddr.split(":")[0];
    int replicaPort = Integer.parseInt(replicaAddr.split(":")[1]);
    TTransport transport = new TSocket(replicaHost, replicaPort);
    try {
        transport.open();
    } catch (TTransportException e) {
        throw new RuntimeException(e);
    }
    TProtocol protocol = new TBinaryProtocol(transport);
    c = new FuseOps.Client(protocol);
    out = new BufferedWriter(new FileWriter(new File(logPrefix + this.id)));
}

/**
 * 启动Thrift服务
 * 
 * @exception STException
 *                异常
 */
public void start() throws STException {
    if (server == null) {
        LOG.debug("Server configuration : " + configuration);
        // 不允许使用1024以内的端口.
        if (!ServerUtil.isSocketPort(serverPort))
            throw new STException("Server initialize failed ! Port range must is 1025 ~ 65535 . Your port is : "
                    + serverPort + " .");
        try {
            initServer(getTransport());
        } catch (TTransportException e) {
            throw new STException("Server initialize failed !", e);
        }
    }
    // 如果服务没有启动,则自动启动服务.
    if (server != null) {
        if (server.isServing()) {
            LOG.info("Server already run .");
            return;
        }
        server.serve();
        LOG.info("Starting server ...... (name : " + serverName + " , port : " + serverPort + ")");
    } else {
        LOG.warn("Startup server failed !");
    }
}

private void reConnect(TTransportException tte) {
    Bgp bgpConfig = bgpConfigSupplier.get();
    if (bgpConfig != null) {
        LOG.error("Received TTransportException, while configuring qthriftd, goind for Disconnect/Connect "
                        + " Host: {}, Port: {}", bgpConfig.getConfigServer().getHost().getValue(),
                bgpConfig.getConfigServer().getPort().intValue());
        disconnect();
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            LOG.error("Exception wile reconnecting ", e);
        }
        connect(bgpConfig.getConfigServer().getHost().getValue(),
                bgpConfig.getConfigServer().getPort().intValue());
    } else {
        LOG.error("Unable to send commands to thrift and fetch bgp configuration", tte);
    }
}

private void readFrame() throws TTransportException {
  transport_.readAll(i32buf, 0, 4);
  int size = decodeFrameSize(i32buf);

  if (size < 0) {
    throw new TTransportException("Read a negative frame size (" + size + ")!");
  }

  if (size > maxLength_) {
    throw new TTransportException("Frame size (" + size + ") larger than max length (" + maxLength_ + ")!");
  }

  byte[] buff = new byte[size];
  transport_.readAll(buff, 0, size);
  readBuffer_.reset(buff);
}

/**
 * Creates a client based on the TypeToken pointing to a Thrift interface class.
 *
 * This method uses black magic, based on the structure and names of generated
 * cc.ecl.action.thrift classes. If that changes, this here breaks.
 *
 * @throws TTransportException    if transport creation fails (e.g. timeout)
 * @throws ThriftFactoryException if factory creation fails (probably bad ifaceType parameter,
 *                                or Thrift internals changed)
 */
@Override
public <IFaceT> IFaceT initializeClient(String url, TypeToken<IFaceT> ifaceType) throws
        TTransportException, ThriftFactoryException {

    // assume same class loader
    ClassLoader classLoader = ifaceType.getRawType().getClassLoader();
    String factoryName = ifaceType.toString().replace("Iface", "Client$Factory");

    try {

        if (!clientFactoryMappings.containsKey(ifaceType)) {
            clientFactoryMappings.put(ifaceType, (TServiceClientFactory<?>) classLoader
                    .loadClass(factoryName).newInstance());
        }
    } catch (Exception e) {
        throw new RuntimeException("IllegalAccessException while initializing: " +
                factoryName, e);
    }

    return (IFaceT) clientFactoryMappings.get(ifaceType).getClient(requestTransport(url));
}

@Override
public int read(byte[] buf, int off, int len) throws TTransportException {
    try {
        return transport.read(buf, off, len);
    } catch (TTransportException ex) {
        this.onException(ex);
        throw ex;
    } catch (RuntimeException re) {
        this.onException(re);
        throw re;
    }
}

@Override
public int readAll(byte[] buf, int off, int len) throws TTransportException {
    try {
        return transport.readAll(buf, off, len);
    } catch (TTransportException ex) {
        this.onException(ex);
        throw ex;
    } catch (RuntimeException re) {
        this.onException(re);
        throw re;
    }
}

private void callSuperClassOpen() throws FlumeException {
  try {
    super.open();
  } catch (TTransportException e) {
    throw new FlumeException("Failed to open SASL transport", e);
  }
}

@BeforeClass
public static void startServer() throws InterruptedException, TTransportException, ConfigurationException, IOException, URISyntaxException  {
    if (! started) {
        EmbeddedCassandraServerHelper.startEmbeddedCassandra(CASSANDRA_UNIT_RANDOM_PORT_YAML, CASSANDRA_TIMEOUT);
        Cluster cluster = new Cluster.Builder().addContactPoints("127.0.0.1").withPort(getNativeTransportPort()).build();
        Session session = cluster.connect();
        String createQuery = "CREATE KEYSPACE " + CASSANDRA_UNIT_KEYSPACE + " WITH replication={'class' : 'SimpleStrategy', 'replication_factor':1}";
        session.execute(createQuery);
        String useKeyspaceQuery = "USE " + CASSANDRA_UNIT_KEYSPACE;
        session.execute(useKeyspaceQuery);
        CQLDataLoader dataLoader = new CQLDataLoader(session);
        applyScripts(dataLoader, "config/cql/changelog/", "*.cql");
        started = true;
    }
}