@Override
public PhoenixTransactionClient getTransactionClient(Configuration config, ConnectionInfo connectionInfo) throws SQLException{
    if (transactionManager == null) {
        try {
            HBaseOmidClientConfiguration clientConf = new HBaseOmidClientConfiguration();
            clientConf.setConflictAnalysisLevel(OmidClientConfiguration.ConflictDetectionLevel.ROW);
            transactionManager = (HBaseTransactionManager) HBaseTransactionManager.newInstance(clientConf);
        } catch (IOException | InterruptedException e) {
            throw new SQLExceptionInfo.Builder(
                    SQLExceptionCode.TRANSACTION_FAILED)
                    .setMessage(e.getMessage()).setRootCause(e).build()
                    .buildException();
        }
    }

    return new OmidTransactionClient(transactionManager);
}
 

@BeforeMethod
public void beforeMethod() throws Exception {

    TSOServerConfig tsoConfig = new TSOServerConfig();
    tsoConfig.setLowLatency(true);
    tsoConfig.setConflictMapSize(1000);
    tsoConfig.setPort(TSO_SERVER_PORT);
    tsoConfig.setNumConcurrentCTWriters(2);
    Module tsoServerMockModule = new TSOMockModule(tsoConfig);
    Injector injector = Guice.createInjector(tsoServerMockModule);

    LOG.info("==================================================================================================");
    LOG.info("======================================= Init TSO Server ==========================================");
    LOG.info("==================================================================================================");

    tsoServer = injector.getInstance(TSOServer.class);
    tsoServer.startAndWait();
    TestUtils.waitForSocketListening(TSO_SERVER_HOST, TSO_SERVER_PORT, 100);

    LOG.info("==================================================================================================");
    LOG.info("===================================== TSO Server Initialized =====================================");
    LOG.info("==================================================================================================");

    pausableTSOracle = (PausableTimestampOracle) injector.getInstance(TimestampOracle.class);
    commitTable = injector.getInstance(CommitTable.class);

    OmidClientConfiguration tsoClientConf = new OmidClientConfiguration();
    tsoClientConf.setConnectionString(TSO_SERVER_HOST + ":" + TSO_SERVER_PORT);

    this.tsoClientConf = tsoClientConf;
    commitTable = injector.getInstance(CommitTable.class);
    replyProcessor = injector.getInstance(ReplyProcessor.class);
}
 

@Override
public OmidClientConfiguration.ConflictDetectionLevel getConflictDetectionLevel() {
    return null;
}
 

public OmidClientConfiguration getOmidClientConfiguration() {
    return omidClientConfiguration;
}
 

public void setOmidClientConfiguration(OmidClientConfiguration omidClientConfiguration) {
    this.omidClientConfiguration = omidClientConfiguration;
}
 

public OmidClientConfiguration.ConnType getConnectionType() {
    return omidClientConfiguration.getConnectionType();
}
 

public void setConnectionType(OmidClientConfiguration.ConnType connectionType) {
    omidClientConfiguration.setConnectionType(connectionType);
}
 

public OmidClientConfiguration getOmidClientConfiguration() {
    return omidClientConfiguration;
}
 

public void setOmidClientConfiguration(OmidClientConfiguration omidClientConfiguration) {
    this.omidClientConfiguration = omidClientConfiguration;
}
 

RawTxRunner(final TSOServerBenchmarkConfig expConfig) throws IOException, InterruptedException {

        // Injector configuration
        List<Module> guiceModules = new ArrayList<>();
        guiceModules.add(new Module() {
            @Override
            public void configure(Binder binder) {
                binder.bind(MetricsRegistry.class).toInstance(expConfig.getMetrics());
            }
        });
        guiceModules.add(expConfig.getCommitTableStoreModule());
        Injector injector = Guice.createInjector(guiceModules);

        // Tx Runner config
        this.writesetSize = expConfig.getWritesetSize();
        this.fixedWriteSetSize = expConfig.isFixedWritesetSize();
        this.commitDelayInMs = expConfig.getCommitDelayInMs();
        this.percentageOfReadOnlyTxs = expConfig.getPercentageOfReadOnlyTxs();
        this.cellIdGenerator = expConfig.getCellIdGenerator();
        this.randomGen = new Random(System.currentTimeMillis() * txRunnerId); // to make it channel dependent

        int txRateInReqPerSec = expConfig.getTxRateInRequestPerSecond();
        long warmUpPeriodInSecs = expConfig.getWarmUpPeriodInSecs();

        LOG.info("TxRunner-{} [ Tx Rate (Req per Sec) -> {} ]", txRunnerId, txRateInReqPerSec);
        LOG.info("TxRunner-{} [ Warm Up Period -> {} Secs ]", txRunnerId, warmUpPeriodInSecs);
        LOG.info("TxRunner-{} [ Cell Id Distribution Generator -> {} ]", txRunnerId, expConfig.getCellIdGenerator().getClass());
        LOG.info("TxRunner-{} [ Max Tx Size -> {} Fixed: {} ]", txRunnerId, writesetSize, fixedWriteSetSize);
        LOG.info("TxRunner-{} [ Commit delay -> {} Ms ]", txRunnerId, commitDelayInMs);
        LOG.info("TxRunner-{} [ % of Read-Only Tx -> {} % ]", txRunnerId, percentageOfReadOnlyTxs);

        // Commit table client initialization
        CommitTable commitTable = injector.getInstance(CommitTable.class);
        this.commitTableClient = commitTable.getClient();

        // Stat initialization
        MetricsRegistry metrics = injector.getInstance(MetricsRegistry.class);
        String hostName = InetAddress.getLocalHost().getHostName();
        this.timestampTimer = metrics.timer(name("tx_runner", Integer.toString(txRunnerId), hostName, "timestamp"));
        this.commitTimer = metrics.timer(name("tx_runner", Integer.toString(txRunnerId), hostName, "commit"));
        this.abortTimer = metrics.timer(name("tx_runner", Integer.toString(txRunnerId), hostName, "abort"));
        this.errorCounter = metrics.counter(name("tx_runner", Integer.toString(txRunnerId), hostName, "errors"));
        LOG.info("TxRunner-{} [ Metrics provider module -> {} ]", txRunnerId, expConfig.getMetrics().getClass());

        // TSO Client initialization
        OmidClientConfiguration tsoClientConf = expConfig.getOmidClientConfiguration();
        this.tsoClient = TSOClient.newInstance(tsoClientConf);
        LOG.info("TxRunner-{} [ Connection Type {}/Connection String {} ]", txRunnerId,
                 tsoClientConf.getConnectionType(), tsoClientConf.getConnectionString());

        // Limiter for configured request per second
        this.rateLimiter = RateLimiter.create((double) txRateInReqPerSec, warmUpPeriodInSecs, TimeUnit.SECONDS);
    }
 

@Override
public PhoenixTransactionService getTransactionService(Configuration config, ConnectionInfo connectionInfo, int port) throws  SQLException{
    TSOServerConfig tsoConfig = new TSOServerConfig();
    TSOServer tso;

    tsoConfig.setPort(port);
    tsoConfig.setConflictMapSize(config.getInt(OMID_TSO_CONFLICT_MAP_SIZE, DEFAULT_OMID_TSO_CONFLICT_MAP_SIZE));
    tsoConfig.setTimestampType(config.get(OMID_TSO_TIMESTAMP_TYPE, DEFAULT_OMID_TSO_TIMESTAMP_TYPE));
    tsoConfig.setWaitStrategy(WAIT_STRATEGY.LOW_CPU.toString());

    Injector injector = Guice.createInjector(new TSOMockModule(tsoConfig));
    tso = injector.getInstance(TSOServer.class);
    tso.startAndWait();

    OmidClientConfiguration clientConfig = new OmidClientConfiguration();
    clientConfig.setConnectionString("localhost:" + port);
    clientConfig.setConflictAnalysisLevel(OmidClientConfiguration.ConflictDetectionLevel.ROW);

    InMemoryCommitTable commitTable = (InMemoryCommitTable) injector.getInstance(CommitTable.class);

    try {
        // Create the associated Handler
        TSOClient client = TSOClient.newInstance(clientConfig);

        HBaseOmidClientConfiguration clientConf = new HBaseOmidClientConfiguration();
        clientConf.setConnectionString("localhost:" + port);
        clientConf.setConflictAnalysisLevel(OmidClientConfiguration.ConflictDetectionLevel.ROW);
        clientConf.setHBaseConfiguration(config);
        commitTableClient = commitTable.getClient();
        commitTableWriter = commitTable.getWriter();
        transactionManager = HBaseTransactionManager.builder(clientConf)
                .commitTableClient(commitTableClient)
                .commitTableWriter(commitTableWriter)
                .tsoClient(client).build();
    } catch (IOException | InterruptedException e) {
        throw new SQLExceptionInfo.Builder(
                SQLExceptionCode.TRANSACTION_FAILED)
                .setMessage(e.getMessage()).setRootCause(e).build()
                .buildException();
    }

    return new OmidTransactionService(tso, transactionManager);
}
 

@Override
public void setConflictDetectionLevel(OmidClientConfiguration.ConflictDetectionLevel conflictDetectionLevel) {

}