[TRAFODION-2129] Trafodion to avoid use of deprecated HBase APIs/Classes References to the deprecated classes HConnection, HConnectionManager and HBaseAdmin are reworked to use Connection, ConnectionFactory and Admin classes respectively.
A connection object is created and it is passed around to ensure that the same connection object is used within the process. Connection object is thread safe. Admin object is created and closed after its use immediately. Creation of admin object is lightweight and the object is not thread safe Also fixed the random hang at the time of rollback transaction waiting for the response from the TM process. The TM process hung because it assumed that RS didn't respond back when there is an exception in DDL rollback or end-point coprocessor call. Project: http://git-wip-us.apache.org/repos/asf/incubator-trafodion/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafodion/commit/7fe96649 Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafodion/tree/7fe96649 Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafodion/diff/7fe96649 Branch: refs/heads/master Commit: 7fe96649f0155bfcc9a6d86328658ac17786258e Parents: 2d7a91f Author: selvaganesang <selva.govindara...@esgyn.com> Authored: Thu Jul 28 21:27:27 2016 +0000 Committer: selvaganesang <se...@apache.org> Committed: Tue Aug 2 18:18:48 2016 +0000 ---------------------------------------------------------------------- .../hbase/client/ClientScanner98.java.tmpl | 24 +- .../client/TrafParallelClientScanner.java.tmpl | 8 +- .../CommitUnsuccessfulException.java | 3 +- .../HBaseBackedTransactionLogger.java | 44 +-- .../OutOfOrderProtocolException.java | 3 +- .../hbase/client/transactional/RMInterface.java | 26 +- .../transactional/SsccTransactionalTable.java | 30 +- .../SsccUpdateConflictException.java | 3 +- .../hbase/client/transactional/TmDDL.java | 26 +- .../transactional/TransactionManager.java | 342 +++++++------------ .../TransactionalAggregationClient.java | 19 +- .../transactional/TransactionalTable.java | 35 +- .../transactional/TransactionalTableClient.java | 3 +- .../transactional/UnsuccessfulDDLException.java | 3 +- .../trafodion/dtm/HBaseAuditControlPoint.java | 23 +- .../java/org/trafodion/dtm/HBaseTxClient.java | 41 ++- .../java/org/trafodion/dtm/TmAuditTlog.java | 52 +-- .../main/java/org/trafodion/dtm/TrafInfo.java | 13 +- core/sql/executor/ExExeUtilLoad.cpp | 12 - core/sql/executor/HBaseClient_JNI.cpp | 44 --- core/sql/executor/HBaseClient_JNI.h | 5 +- core/sql/executor/SequenceFileReader.cpp | 26 -- core/sql/executor/SequenceFileReader.h | 2 - core/sql/regress/privs1/DIFF133.KNOWN | 39 +-- .../java/org/trafodion/sql/HBaseClient.java | 126 +++---- .../java/org/trafodion/sql/HBulkLoadClient.java | 62 ++-- .../java/org/trafodion/sql/HTableClient.java | 36 +- .../org/trafodion/sql/SequenceFileWriter.java | 40 +-- .../java/org/trafodion/sql/TrafRegionStats.java | 16 +- 29 files changed, 424 insertions(+), 682 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/7fe96649/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/ClientScanner98.java.tmpl ---------------------------------------------------------------------- diff --git a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/ClientScanner98.java.tmpl b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/ClientScanner98.java.tmpl index 4d9fad9..9287814 100644 --- a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/ClientScanner98.java.tmpl +++ b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/ClientScanner98.java.tmpl @@ -62,7 +62,7 @@ public class ClientScanner98 extends AbstractClientScanner { // Keep lastResult returned successfully in case we have to reset scanner. protected Result lastResult = null; protected final long maxScannerResultSize; - private final HConnection connection; + private final Connection connection; private final TableName tableName; protected final int scannerTimeout; protected boolean scanMetricsPublished = false; @@ -70,7 +70,7 @@ public class ClientScanner98 extends AbstractClientScanner { protected RpcControllerFactory rpcControllerFactory; /** - * Create a new ClientScanner for the specified table. An HConnection will be + * Create a new ClientScanner for the specified table. An Connection will be * retrieved using the passed Configuration. * Note that the passed {@link Scan}'s start row maybe changed changed. * @@ -79,20 +79,24 @@ public class ClientScanner98 extends AbstractClientScanner { * @param tableName The table that we wish to scan * @throws IOException */ +/* @Deprecated public ClientScanner98(final Configuration conf, final Scan scan, final TableName tableName) throws IOException { - this(conf, scan, tableName, HConnectionManager.getConnection(conf)); + this(conf, scan, tableName, ConnectionFactory.createConnection(conf)); } +*/ /** * @deprecated Use {@link #ClientScanner(Configuration, Scan, TableName)} */ +/* @Deprecated public ClientScanner98(final Configuration conf, final Scan scan, final byte [] tableName) throws IOException { this(conf, scan, TableName.valueOf(tableName)); } +*/ /** @@ -106,30 +110,30 @@ public class ClientScanner98 extends AbstractClientScanner { * @throws IOException */ public ClientScanner98(final Configuration conf, final Scan scan, final TableName tableName, - HConnection connection) throws IOException { + Connection connection) throws IOException { this(conf, scan, tableName, connection, RpcRetryingCallerFactory.instantiate(conf), RpcControllerFactory.instantiate(conf)); } /** - * @deprecated Use {@link #ClientScanner(Configuration, Scan, TableName, HConnection)} + * @deprecated Use {@link #ClientScanner(Configuration, Scan, TableName, Connection)} */ @Deprecated public ClientScanner98(final Configuration conf, final Scan scan, final byte [] tableName, - HConnection connection) throws IOException { + Connection connection) throws IOException { this(conf, scan, TableName.valueOf(tableName), connection, new RpcRetryingCallerFactory(conf), RpcControllerFactory.instantiate(conf)); } /** * @deprecated Use - * {@link #ClientScanner(Configuration, Scan, TableName, HConnection, + * {@link #ClientScanner(Configuration, Scan, TableName, Connection, * RpcRetryingCallerFactory, RpcControllerFactory)} * instead */ @Deprecated public ClientScanner98(final Configuration conf, final Scan scan, final TableName tableName, - HConnection connection, RpcRetryingCallerFactory rpcFactory) throws IOException { + Connection connection, RpcRetryingCallerFactory rpcFactory) throws IOException { this(conf, scan, tableName, connection, rpcFactory, RpcControllerFactory.instantiate(conf)); } @@ -143,7 +147,7 @@ public class ClientScanner98 extends AbstractClientScanner { * @throws IOException */ public ClientScanner98(final Configuration conf, final Scan scan, final TableName tableName, - HConnection connection, RpcRetryingCallerFactory rpcFactory, + Connection connection, RpcRetryingCallerFactory rpcFactory, RpcControllerFactory controllerFactory) throws IOException { if (LOG.isTraceEnabled()) { LOG.trace("Scan table=" + tableName @@ -188,7 +192,7 @@ public class ClientScanner98 extends AbstractClientScanner { nextScanner(this.caching, false); } - protected HConnection getConnection() { + protected Connection getConnection() { return this.connection; } http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/7fe96649/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/TrafParallelClientScanner.java.tmpl ---------------------------------------------------------------------- diff --git a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/TrafParallelClientScanner.java.tmpl b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/TrafParallelClientScanner.java.tmpl index 3ff9a81..7b61a8d 100644 --- a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/TrafParallelClientScanner.java.tmpl +++ b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/TrafParallelClientScanner.java.tmpl @@ -81,7 +81,7 @@ public class TrafParallelClientScanner extends AbstractClientScanner implements * (should be larger than the individual scans' caching setting) * @throws IOException */ - public TrafParallelClientScanner(final HConnection connection, final Collection<Scan> scans, + public TrafParallelClientScanner(final Connection connection, final Collection<Scan> scans, final TableName tableName, final int bufferSize) throws IOException { createDefaultPool(scans.size()); SingleReaderMultiWriterQueue rw = new SingleReaderMultiWriterQueue(bufferSize, scans.size()); @@ -100,7 +100,7 @@ public class TrafParallelClientScanner extends AbstractClientScanner implements * @param parallelScaling The scaling factor to use. if between 0 and 1, percent of regions run in parallel, if > 1 fixed number of thread run in parallel * @throws IOException */ - public TrafParallelClientScanner(final HConnection connection, final Scan scan, + public TrafParallelClientScanner(final Connection connection, final Scan scan, final TableName tableName, final float parallelScaling) throws IOException { if (parallelScaling <= 0) throw new IllegalArgumentException("scalingFactor must > 0"); @@ -288,10 +288,10 @@ public class TrafParallelClientScanner extends AbstractClientScanner implements protected class ScanTask implements Runnable { private Scan s; private ResultWriter writer; - private HConnection connection; + private Connection connection; private TableName tableName; private int taskNb; - public ScanTask(Scan s, HConnection connection, TableName tableName, ResultWriter writer, int taskNb) { + public ScanTask(Scan s, Connection connection, TableName tableName, ResultWriter writer, int taskNb) { this.s = s; this.connection = connection; this.tableName = tableName; http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/7fe96649/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/CommitUnsuccessfulException.java ---------------------------------------------------------------------- diff --git a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/CommitUnsuccessfulException.java b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/CommitUnsuccessfulException.java index 3b49c6b..2991b5a 100644 --- a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/CommitUnsuccessfulException.java +++ b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/CommitUnsuccessfulException.java @@ -22,11 +22,12 @@ **/ package org.apache.hadoop.hbase.client.transactional; +import java.io.IOException; /** Thrown when a transaction cannot be committed. * */ -public class CommitUnsuccessfulException extends Exception { +public class CommitUnsuccessfulException extends IOException { private static final long serialVersionUID = 7062921444531109202L; http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/7fe96649/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/HBaseBackedTransactionLogger.java ---------------------------------------------------------------------- diff --git a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/HBaseBackedTransactionLogger.java b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/HBaseBackedTransactionLogger.java index 5561aa1..cbf9e9a 100644 --- a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/HBaseBackedTransactionLogger.java +++ b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/HBaseBackedTransactionLogger.java @@ -26,14 +26,14 @@ package org.apache.hadoop.hbase.client.transactional; import java.io.IOException; import java.util.Random; -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HColumnDescriptor; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.HTableDescriptor; import org.apache.hadoop.hbase.client.Delete; import org.apache.hadoop.hbase.client.Get; -import org.apache.hadoop.hbase.client.HBaseAdmin; +import org.apache.hadoop.hbase.client.Admin; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.Connection; import org.apache.hadoop.hbase.client.HTableInterface; import org.apache.hadoop.hbase.client.HTablePool; import org.apache.hadoop.hbase.client.Put; @@ -45,35 +45,28 @@ public class HBaseBackedTransactionLogger implements TransactionLogger { /** The name of the transaction status table. */ public static final String TABLE_NAME = "__GLOBAL_TRX_LOG__"; - private static final byte[] INFO_FAMILY = Bytes.toBytes("Info"); + private static final byte[] INFO_FAMILY = Bytes.toBytes("Info"); /** * Column which holds the transaction status. */ private static final byte[] STATUS_QUALIFIER = Bytes.toBytes("Status"); - /** - * Create the global transaction table. - * - * @throws IOException - */ - public static void createTable() throws IOException { - createTable(HBaseConfiguration.create()); - } /** * Create the global transaction table with the given configuration. * - * @param conf + * @param Connection * @throws IOException */ - public static void createTable(final Configuration conf) throws IOException { + public static void createTable(Connection connection) throws IOException { HTableDescriptor tableDesc = new HTableDescriptor(TABLE_NAME); tableDesc.addFamily(new HColumnDescriptor(INFO_FAMILY)); - HBaseAdmin admin = new HBaseAdmin(conf); - if (!admin.tableExists(TABLE_NAME)) { + Admin admin = connection.getAdmin(); + if (! admin.tableExists(TableName.valueOf(TABLE_NAME))) { admin.createTable(tableDesc); } + admin.close(); } private Random random = new Random(); @@ -90,20 +83,17 @@ public class HBaseBackedTransactionLogger implements TransactionLogger { tablePool.putTable(t); } - public HBaseBackedTransactionLogger() throws IOException { - initTable(HBaseConfiguration.create()); - } - - public HBaseBackedTransactionLogger(final Configuration conf) throws IOException { - initTable(conf); + public HBaseBackedTransactionLogger(Connection connection) throws IOException { + initTable(connection); } - private void initTable(final Configuration conf) throws IOException { - HBaseAdmin admin = new HBaseAdmin(conf); - - if (!admin.tableExists(TABLE_NAME)) { + private void initTable(Connection connection) throws IOException { + boolean retcode ; + Admin admin = connection.getAdmin(); + retcode = (!admin.tableExists(TableName.valueOf(TABLE_NAME))); + admin.close(); + if (retcode) throw new RuntimeException("Table not created. Call createTable() first"); - } } public long createNewTransactionLog() { http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/7fe96649/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/OutOfOrderProtocolException.java ---------------------------------------------------------------------- diff --git a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/OutOfOrderProtocolException.java b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/OutOfOrderProtocolException.java index 166ec03..84bf6f2 100644 --- a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/OutOfOrderProtocolException.java +++ b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/OutOfOrderProtocolException.java @@ -22,12 +22,13 @@ **/ package org.apache.hadoop.hbase.client.transactional; +import java.io.IOException; /** Thrown when the TrxRegionEndpoint openScanner, performScan, * closeScanner protocol is not being followed in the correct order. */ -public class OutOfOrderProtocolException extends Exception { +public class OutOfOrderProtocolException extends IOException { private static final long serialVersionUID = 1712128652088625482L; http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/7fe96649/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/RMInterface.java ---------------------------------------------------------------------- diff --git a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/RMInterface.java b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/RMInterface.java index e1740c4..4506ebc 100644 --- a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/RMInterface.java +++ b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/RMInterface.java @@ -34,11 +34,10 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.commons.codec.binary.Hex; import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HTableDescriptor; import org.apache.hadoop.hbase.HRegionInfo; import org.apache.hadoop.hbase.HRegionLocation; -import org.apache.hadoop.hbase.client.HConnection; +import org.apache.hadoop.hbase.client.Connection; import org.apache.hadoop.hbase.client.HTable; import org.apache.hadoop.hbase.client.Delete; import org.apache.hadoop.hbase.client.Get; @@ -91,6 +90,7 @@ public class RMInterface { public AlgorithmType TRANSACTION_ALGORITHM; static Map<Long, Set<RMInterface>> mapRMsPerTransaction = new HashMap<Long, Set<RMInterface>>(); private TransactionalTableClient ttable = null; + private Connection connection; static { System.loadLibrary("stmlib"); } @@ -114,8 +114,9 @@ public class RMInterface { private AlgorithmType transactionAlgorithm; - public RMInterface(final String tableName) throws IOException { + public RMInterface(final String tableName, Connection connection) throws IOException { //super(conf, Bytes.toBytes(tableName)); + this.connection = connection; transactionAlgorithm = AlgorithmType.MVCC; String envset = System.getenv("TM_USE_SSCC"); if( envset != null) @@ -124,23 +125,18 @@ public class RMInterface { } if( transactionAlgorithm == AlgorithmType.MVCC) //MVCC { - ttable = new TransactionalTable(Bytes.toBytes(tableName)); + ttable = new TransactionalTable(Bytes.toBytes(tableName), connection); } else if(transactionAlgorithm == AlgorithmType.SSCC) { - ttable = new SsccTransactionalTable( Bytes.toBytes(tableName)); - } - - try { - idServer = new IdTm(false); - } - catch (Exception e){ - LOG.error("RMInterface: Exception creating new IdTm: " + e); + ttable = new SsccTransactionalTable( Bytes.toBytes(tableName), connection); } + idServer = new IdTm(false); if (LOG.isTraceEnabled()) LOG.trace("RMInterface constructor exit"); } - public RMInterface() throws IOException { + public RMInterface(Connection connection) throws IOException { + this.connection = connection; } @@ -369,10 +365,6 @@ public class RMInterface { public void flushCommits() throws IOException { ttable.flushCommits(); } - public HConnection getConnection() - { - return ttable.getConnection(); - } public byte[][] getEndKeys() throws IOException { http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/7fe96649/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/SsccTransactionalTable.java ---------------------------------------------------------------------- diff --git a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/SsccTransactionalTable.java b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/SsccTransactionalTable.java index 3735b63..9c76bcc 100644 --- a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/SsccTransactionalTable.java +++ b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/SsccTransactionalTable.java @@ -37,12 +37,11 @@ import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException; import org.apache.commons.codec.binary.Hex; -import org.apache.hadoop.hbase.client.HConnection; +import org.apache.hadoop.hbase.client.Connection; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.Cell; -import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HRegionLocation; import org.apache.hadoop.hbase.KeyValue; import org.apache.hadoop.hbase.KeyValueUtil; @@ -61,7 +60,6 @@ import org.apache.hadoop.hbase.protobuf.ProtobufUtil; import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto; import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto.MutationType; import org.apache.hadoop.hbase.regionserver.transactional.SingleVersionDeleteNotSupported; -import org.apache.hadoop.hbase.client.HConnectionManager; import org.apache.hadoop.hbase.util.Bytes; import com.google.protobuf.ByteString; @@ -119,8 +117,7 @@ public class SsccTransactionalTable extends HTable implements TransactionalTable private RpcRetryingCallerFactory rpcCallerFactory; private RpcControllerFactory rpcControllerFactory; */ - static private HConnection connection = null; - static Configuration config = HBaseConfiguration.create(); + static private Connection connection = null; static ExecutorService threadPool; private static final int STATEFUL_UPDATE_OK = 1; @@ -128,24 +125,13 @@ public class SsccTransactionalTable extends HTable implements TransactionalTable private static final int STATELESS_UPDATE_OK = 3; private static final int STATELESS_UPDATE_CONFLICT = 5; - static { - config.set("hbase.hregion.impl", "org.apache.hadoop.hbase.regionserver.transactional.TransactionalRegion"); - try { - connection = HConnectionManager.createConnection(config); - } - catch (IOException ioe) { - LOG.error("Exception on TransactionTable anonymous static method. IOException while creating HConnection"); - } - threadPool = Executors.newCachedThreadPool(); - } - /** * @param conf * @param tableName * @throws IOException */ - public SsccTransactionalTable( final String tableName) throws IOException { - this(Bytes.toBytes(tableName)); + public SsccTransactionalTable( final String tableName, Connection connection) throws IOException { + this(Bytes.toBytes(tableName), connection); } /** @@ -153,8 +139,9 @@ public class SsccTransactionalTable extends HTable implements TransactionalTable * @param tableName * @throws IOException */ - public SsccTransactionalTable( final byte[] tableName) throws IOException { + public SsccTransactionalTable( final byte[] tableName, Connection connection) throws IOException { super( tableName,connection, threadPool); + this.connection = connection; } private void addLocation(final TransactionState transactionState, HRegionLocation location) { @@ -688,10 +675,7 @@ if (LOG.isTraceEnabled()) LOG.trace("checkAndPut, seting request startid: " + tr public void flushCommits() throws IOException { super.flushCommits(); } - public HConnection getConnection() - { - return super.getConnection(); - } + public byte[][] getEndKeys() throws IOException { http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/7fe96649/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/SsccUpdateConflictException.java ---------------------------------------------------------------------- diff --git a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/SsccUpdateConflictException.java b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/SsccUpdateConflictException.java index 2db3638..4d80d1e 100644 --- a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/SsccUpdateConflictException.java +++ b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/SsccUpdateConflictException.java @@ -22,11 +22,12 @@ **/ package org.apache.hadoop.hbase.client.transactional; +import java.io.IOException; /** Thrown when a transaction cannot be committed. * */ -public class SsccUpdateConflictException extends Exception { +public class SsccUpdateConflictException extends IOException { private static final long serialVersionUID = 7062921444531109202L; http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/7fe96649/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TmDDL.java ---------------------------------------------------------------------- diff --git a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TmDDL.java b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TmDDL.java index 514b5bd..83eb945 100644 --- a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TmDDL.java +++ b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TmDDL.java @@ -36,11 +36,9 @@ import java.io.ByteArrayInputStream; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.hbase.HBaseConfiguration; -import org.apache.hadoop.hbase.client.HBaseAdmin; -import org.apache.hadoop.hbase.client.HConnection; -import org.apache.hadoop.hbase.client.HConnectionManager; -import org.apache.hadoop.hbase.client.HTable; +import org.apache.hadoop.hbase.client.Admin; +import org.apache.hadoop.hbase.client.Connection; +import org.apache.hadoop.hbase.client.Table; import org.apache.hadoop.hbase.client.HTableInterface; import org.apache.hadoop.hbase.HTableDescriptor; import org.apache.hadoop.hbase.client.Get; @@ -57,7 +55,7 @@ import org.apache.hadoop.hbase.util.Bytes; public class TmDDL { static final Log LOG = LogFactory.getLog(TmDDL.class); - private HBaseAdmin hbadmin; + private Connection connection; private Configuration config; private int dtmid; private static final byte[] TDDL_FAMILY = Bytes.toBytes("tddlcf"); @@ -66,29 +64,31 @@ public class TmDDL { private static final byte[] TDDL_TRUNCATE = Bytes.toBytes("truncateList"); private static final byte[] TDDL_STATE = Bytes.toBytes("state"); private static Object tablePutLock; // Lock for synchronizing table.put operations - private static HTable table; + private static Table table; - public TmDDL (Configuration config) throws IOException { + public TmDDL (Configuration config, Connection connection) throws IOException { this.config = config; + this.connection = connection; this.dtmid = Integer.parseInt(config.get("dtmid")); TableName tablename = TableName.valueOf("TRAFODION._DTM_.TDDL"); if (LOG.isTraceEnabled()) LOG.trace("Enter TmDDL constructor for dtmid: " + dtmid); - hbadmin = new HBaseAdmin(config); + Admin admin = connection.getAdmin(); - boolean tDDLTableExists = hbadmin.tableExists(tablename); + boolean tDDLTableExists = admin.tableExists(tablename); - if(tDDLTableExists==false && dtmid ==0) { + if (tDDLTableExists==false && dtmid ==0) { HTableDescriptor desc = new HTableDescriptor(tablename); desc.addFamily(new HColumnDescriptor(TDDL_FAMILY)); - hbadmin.createTable(desc); + admin.createTable(desc); } tablePutLock = new Object(); - table = new HTable(config, tablename); + table = connection.getTable(tablename); + admin.close(); } public void putRow(final long transid, final String Operation, final String tableName) throws IOException { http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/7fe96649/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionManager.java ---------------------------------------------------------------------- diff --git a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionManager.java b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionManager.java index 1ddd080..425802e 100644 --- a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionManager.java +++ b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionManager.java @@ -54,19 +54,16 @@ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.MasterNotRunningException; import org.apache.hadoop.hbase.TableNotFoundException; -import org.apache.hadoop.hbase.TableNotEnabledException; -import org.apache.hadoop.hbase.TableNotDisabledException; import org.apache.hadoop.hbase.DoNotRetryIOException; -import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.HRegionInfo; import org.apache.hadoop.hbase.HRegionLocation; import org.apache.hadoop.hbase.ZooKeeperConnectionException; import org.apache.hadoop.hbase.HTableDescriptor; import org.apache.hadoop.hbase.HColumnDescriptor; -import org.apache.hadoop.hbase.client.HBaseAdmin; -import org.apache.hadoop.hbase.client.HConnection; -import org.apache.hadoop.hbase.client.HConnectionManager; +import org.apache.hadoop.hbase.client.Admin; +import org.apache.hadoop.hbase.client.Connection; +import org.apache.hadoop.hbase.client.ConnectionFactory; import org.apache.hadoop.hbase.client.HTable; import org.apache.hadoop.hbase.client.coprocessor.Batch; import org.apache.hadoop.hbase.client.Durability; @@ -146,13 +143,11 @@ public class TransactionManager { private boolean batchRegionServer = false; private int RETRY_ATTEMPTS; - private final HConnection connection; private final TransactionLogger transactionLogger; private JtaXAResource xAResource; - private HBaseAdmin hbadmin; + private Connection connection; private TmDDL tmDDL; private boolean batchRSMetricsFlag = false; - Configuration config; public static final int HBASE_NAME = 0; public static final int HBASE_MAX_VERSIONS = 1; @@ -202,9 +197,10 @@ public class TransactionManager { } // getInstance to return the singleton object for TransactionManager - public synchronized static TransactionManager getInstance(final Configuration conf) throws ZooKeeperConnectionException, IOException { + public synchronized static TransactionManager getInstance(final Configuration conf, Connection connection) + throws ZooKeeperConnectionException, IOException { if (g_TransactionManager == null) { - g_TransactionManager = new TransactionManager(conf); + g_TransactionManager = new TransactionManager(conf, connection); } return g_TransactionManager; } @@ -236,9 +232,7 @@ public class TransactionManager { } public void init(final TmDDL tmddl) throws IOException { - this.config = HBaseConfiguration.create(); this.tmDDL = tmddl; - hbadmin = new HBaseAdmin(config); } /** @@ -252,7 +246,7 @@ public class TransactionManager { byte[] endKey_orig; byte[] endKey; - TransactionManagerCallable(TransactionState txState, TransactionRegionLocation location, HConnection connection) + TransactionManagerCallable(TransactionState txState, TransactionRegionLocation location, Connection connection) throws IOException { transactionState = txState; this.location = location; @@ -324,6 +318,7 @@ public class TransactionManager { String msg = new String ("ERROR occurred while calling doCommitX coprocessor service in doCommitX for transaction: " + transactionId + " participantNum " + participantNum ); LOG.error(msg, e); + transactionState.requestPendingCountDec(true); throw new DoNotRetryIOException(msg,e); } if(result.size() == 0) { @@ -345,15 +340,7 @@ public class TransactionManager { ArrayList<String> dropList = new ArrayList<String>(); ArrayList<String> truncateList = new ArrayList<String>(); StringBuilder state = new StringBuilder (); - try { - tmDDL.getRow(transactionState.getTransactionId(), state, createList, dropList, truncateList); - } - catch(Exception e){ - LOG.error("doCommitX, exception in tmDDL getRow: " + e); - if(LOG.isTraceEnabled()) LOG.trace("doCommitX, exception in tmDDL getRow:: txID: " + transactionState.getTransactionId()); - state.append("INVALID"); //to avoid processing further down this path. - } - + tmDDL.getRow(transactionState.getTransactionId(), state, createList, dropList, truncateList); if(state.toString().equals("VALID") && dropList.size() > 0) { Iterator<String> di = dropList.iterator(); @@ -510,6 +497,7 @@ public class TransactionManager { } catch (ServiceException se) { String msg = "ERROR occurred while calling doCommitX coprocessor service in doCommitX"; LOG.error(msg + ":", se); + transactionState.requestPendingCountDec(true); throw new DoNotRetryIOException(msg, se); } catch (Throwable e) { String msg = "ERROR occurred while calling doCommitX coprocessor service in doCommitX"; @@ -940,7 +928,8 @@ public class TransactionManager { int retryCount = 0; int retrySleep = TM_SLEEP; - if( TRANSACTION_ALGORITHM == AlgorithmType.MVCC){ + Admin admin = connection.getAdmin(); + if( TRANSACTION_ALGORITHM == AlgorithmType.MVCC) { do { try { @@ -975,6 +964,7 @@ public class TransactionManager { } catch (Throwable t) { String msg = "ERROR occurred while calling doAbortX coprocessor service"; LOG.error(msg, t); + transactionState.requestPendingCountDec(true); throw new DoNotRetryIOException(msg, t); } @@ -1002,6 +992,12 @@ public class TransactionManager { retry = false; } } + catch (UnknownTransactionException ute) { + LOG.error("Got unknown exception in doAbortX by participant " + participantNum + + " for transaction: " + transactionId, ute); + transactionState.requestPendingCountDec(true); + throw ute; + } catch (RetryTransactionException rte) { if (rte.toString().contains("Asked to commit a non-pending transaction ")) { LOG.error(" doCommitX will not retry transaction: " + transactionId , rte); @@ -1011,6 +1007,7 @@ public class TransactionManager { if (retryCount == RETRY_ATTEMPTS) { String errMsg = "Exceeded retry attempts in doAbortX: " + retryCount; LOG.error(errMsg, rte); + transactionState.requestPendingCountDec(true); throw new DoNotRetryIOException(errMsg, rte); } else { @@ -1081,6 +1078,7 @@ public class TransactionManager { } catch (Throwable e) { String msg = "ERROR occurred while calling doAbortX coprocessor service"; LOG.error(msg + ":", e); + transactionState.requestPendingCountDec(true); throw new DoNotRetryIOException(msg,e); } @@ -1104,10 +1102,17 @@ public class TransactionManager { retry = false; } } + catch (UnknownTransactionException ute) { + LOG.error("Got unknown exception in doAbortX by participant " + participantNum + + " for transaction: " + transactionId, ute); + transactionState.requestPendingCountDec(true); + throw ute; + } catch (RetryTransactionException rte) { if (retryCount == RETRY_ATTEMPTS){ String errMsg = new String ("Exceeded retry attempts in doAbortX: " + retryCount + " (Not ingoring)"); LOG.error(errMsg); + transactionState.requestPendingCountDec(true); throw new RollbackUnsuccessfulException(errMsg, rte); } LOG.error("doAbortX participant " + participantNum + " retrying transaction " @@ -1144,6 +1149,7 @@ public class TransactionManager { } while (retryCount < RETRY_ATTEMPTS && retry == true); } + admin.close(); // We have received our reply so decrement outstanding count transactionState.requestPendingCountDec(false); @@ -1420,6 +1426,7 @@ public class TransactionManager { if(retryCount == RETRY_ATTEMPTS){ String errMsg = "Exceeded retry attempts in doAbortX: " + retryCount + " (not ingoring)"; LOG.error(errMsg, rte); + transactionState.requestPendingCountDec(true); throw new RollbackUnsuccessfulException("doAbortX, Batch - coprocessor error", rte); } LOG.error("doAbortX - Batch - participant " + participantNum + " retrying transaction " @@ -1435,10 +1442,6 @@ public class TransactionManager { + "-- location being refreshed : " + location.getRegionInfo().getRegionNameAsString() + " endKey: " + Hex.encodeHexString(location.getRegionInfo().getEndKey()) + " for transaction: " + transactionId); - if(retryCount == RETRY_ATTEMPTS){ - LOG.error("Exceeded retry attempts in doAbortX: " + retryCount + " (ingoring)"); - } - if (LOG.isWarnEnabled()) { LOG.warn("doAbortX - Batch - -- " + table.toString() + " location being refreshed"); LOG.warn("doAbortX - Batch - -- lv_hri: " + lv_hri); @@ -1446,19 +1449,19 @@ public class TransactionManager { } table.getRegionLocation(startKey, true); - if (LOG.isTraceEnabled()) LOG.trace("doAbortX - Batch - -- setting retry, count: " + retryCount); - refresh = false; - retryCount++; - } + if (LOG.isTraceEnabled()) LOG.trace("doAbortX - Batch - -- setting retry, count: " + retryCount); + refresh = false; + retryCount++; + } } while (retryCount < RETRY_ATTEMPTS && retry == true); transactionState.requestPendingCountDec(false); if(LOG.isTraceEnabled()) LOG.trace("doAbortX - Batch -- EXIT txID: " + transactionId); return 0; - } + } public Integer pushRegionEpochX(final TransactionState txState, - final HRegionLocation location, HConnection connection) throws IOException { + final HRegionLocation location, Connection connection) throws IOException { if (LOG.isTraceEnabled()) LOG.trace("pushRegionEpochX -- Entry txState: " + txState + " location: " + location); @@ -1559,9 +1562,9 @@ public class TransactionManager { * @param conf * @throws ZooKeeperConnectionException */ - private TransactionManager(final Configuration conf) throws ZooKeeperConnectionException, IOException { - this(LocalTransactionLogger.getInstance(), conf); - + private TransactionManager(final Configuration conf, Connection connection) throws ZooKeeperConnectionException, IOException { + this(LocalTransactionLogger.getInstance(), conf, connection); + this.connection = connection; int intThreads = 16; String retryAttempts = System.getenv("TMCLIENT_RETRY_ATTEMPTS"); String numThreads = System.getenv("TM_JAVA_THREAD_POOL_SIZE"); @@ -1592,12 +1595,7 @@ public class TransactionManager { if (useSSCC != null) TRANSACTION_ALGORITHM = (Integer.parseInt(useSSCC) == 1) ? AlgorithmType.SSCC :AlgorithmType.MVCC ; - try { - idServer = new IdTm(false); - } - catch (Exception e){ - LOG.error("Exception creating new IdTm: " + e); - } + idServer = new IdTm(false); threadPool = Executors.newFixedThreadPool(intThreads); @@ -1617,14 +1615,16 @@ public class TransactionManager { * @param conf * @throws ZooKeeperConnectionException */ - protected TransactionManager(final TransactionLogger transactionLogger, final Configuration conf) + + protected TransactionManager(final TransactionLogger transactionLogger, final Configuration conf, Connection conn) throws ZooKeeperConnectionException, IOException { this.transactionLogger = transactionLogger; conf.setInt("hbase.client.retries.number", 3); - connection = HConnectionManager.createConnection(conf); + connection = conn; } + /** * Called to start a transaction. * @@ -1934,38 +1934,14 @@ public class TransactionManager { ArrayList<String> dropList = new ArrayList<String>(); ArrayList<String> truncateList = new ArrayList<String>(); StringBuilder state = new StringBuilder (); - try { - tmDDL.getRow(transactionState.getTransactionId(), state, createList, dropList, truncateList); - } - catch(Exception e){ - LOG.error("exception in doPrepare getRow: " + e); - if(LOG.isTraceEnabled()) LOG.trace("exception in doPrepare getRow: txID: " + transactionState.getTransactionId()); - state.append("INVALID"); //to avoid processing further down this path. - commitError = TransactionalReturn.COMMIT_UNSUCCESSFUL; - } - - //Return if error at this point. - if(commitError != 0) - return commitError; - + tmDDL.getRow(transactionState.getTransactionId(), state, createList, dropList, truncateList); if(state.toString().equals("VALID") && dropList.size() > 0) { Iterator<String> di = dropList.iterator(); while (di.hasNext()) { - try { - //physical drop of table from hbase. - disableTable(transactionState, di.next()); - } - catch(Exception e){ - if(LOG.isTraceEnabled()) LOG.trace("exception in doPrepare disableTable: txID: " + transactionState.getTransactionId()); - LOG.error("exception in doCommit, Step : DeleteTable: " + e); - - //Any error at this point should be considered prepareCommit as unsuccessful. - //Retry logic can be added only if it is retryable error: TODO. - commitError = TransactionalReturn.COMMIT_UNSUCCESSFUL; - break; - } + //physical drop of table from hbase. + disableTable(transactionState, di.next()); } } } @@ -2038,8 +2014,7 @@ public class TransactionManager { public void pushRegionEpoch (HTableDescriptor desc, final TransactionState ts) throws IOException { LOG.info("pushRegionEpoch start; transId: " + ts.getTransactionId()); - TransactionalTable ttable1 = new TransactionalTable(Bytes.toBytes(desc.getNameAsString())); - HConnection connection = ttable1.getConnection(); + TransactionalTable ttable1 = new TransactionalTable(Bytes.toBytes(desc.getNameAsString()), connection); long lvTransid = ts.getTransactionId(); RegionLocator rl = connection.getRegionLocator(desc.getTableName()); List<HRegionLocation> regionList = rl.getAllRegionLocations(); @@ -2051,12 +2026,11 @@ public class TransactionManager { int result = 0; for (HRegionLocation location : regionList) { final byte[] regionName = location.getRegionInfo().getRegionName(); - final HConnection lv_connection = connection; final TransactionRegionLocation lv_location = new TransactionRegionLocation(location.getRegionInfo(), location.getServerName()); - compPool.submit(new TransactionManagerCallable(ts, lv_location, lv_connection) { + compPool.submit(new TransactionManagerCallable(ts, lv_location, connection) { public Integer call() throws IOException { - return pushRegionEpochX(ts, lv_location, lv_connection); + return pushRegionEpochX(ts, lv_location, connection); } }); boolean loopExit = false; @@ -2256,7 +2230,7 @@ public class TransactionManager { tmDDL.getRow(transactionState.getTransactionId(), state, createList, dropList, truncateList); retry = false; } - catch(Exception e){ + catch(IOException e){ LOG.error("Exception in doCommitDDL, Step: getRow. txID: " + transactionState.getTransactionId() + "Exception: " + e); if(retryCount == RETRY_ATTEMPTS) @@ -2280,7 +2254,7 @@ public class TransactionManager { } } while (retryCount < RETRY_ATTEMPTS && retry == true); - if(state.toString().equals("VALID") && dropList.size() > 0) + if (state.toString().equals("VALID") && dropList.size() > 0) { Iterator<String> di = dropList.iterator(); while (di.hasNext()) @@ -2303,7 +2277,7 @@ public class TransactionManager { if(LOG.isTraceEnabled()) LOG.trace(" TableNotFoundException exception in doCommitDDL deleteTable, Continuing: txID: " + transactionState.getTransactionId()); retry = false; } - catch(Exception e){ + catch(IOException e){ LOG.error("Fatal exception in doCommitDDL, Step : DeleteTable: TxID:" + transactionState.getTransactionId() + "Exception: " + e); if(retryCount == RETRY_ATTEMPTS) @@ -2340,7 +2314,7 @@ public class TransactionManager { tmDDL.deleteRow(transactionState.getTransactionId()); retry = false; } - catch(Exception e) + catch (IOException e) { LOG.error("Fatal Exception in doCommitDDL, Step: deleteRow. txID: " + transactionState.getTransactionId() + "Exception: " + e); @@ -2423,7 +2397,6 @@ public class TransactionManager { if (transactionState.getRegionsToIgnore().contains(location)) { continue; } - //try { loopCount++; final int participantNum = loopCount; final byte[] regionName = location.getRegionInfo().getRegionName(); @@ -2436,27 +2409,13 @@ public class TransactionManager { return doAbortX(regionName, transactionState.getTransactionId(), participantNum, location.isTableRecodedDropped()); } }); -/* - } catch (Exception e) { - LOG.error("exception in abort: " + e); - } -*/ - /* - } catch (UnknownTransactionException e) { - LOG.error("exception in abort: " + e); - LOG.info("Got unknown transaction exception during abort. Transaction: [" - + transactionState.getTransactionId() + "], region: [" - + location.getRegionInfo().getRegionNameAsString() + "]. Ignoring."); - } catch (NotServingRegionException e) { - LOG.info("Got NSRE during abort. Transaction: [" + transactionState.getTransactionId() + "], region: [" - + location.getRegionInfo().getRegionNameAsString() + "]. Ignoring."); - } - */ } // all requests sent at this point, can record the count transactionState.completeSendInvoke(loopCount); } + + CommitUnsuccessfulException savedCue = null; //if DDL is involved with this transaction, need to unwind it. if(transactionState.hasDDLTx()) @@ -2464,28 +2423,30 @@ public class TransactionManager { //First wait for abort requests sent to all regions is received back. //This TM thread gets SUSPENDED until all abort threads complete!!! - try{ + boolean loopExit = false; + do + { + try { transactionState.completeRequest(); - } - catch(Exception e){ - LOG.error("Exception in abort() completeRequest. txID: " + transactionState.getTransactionId() + "Exception: " + e); - //return; //Do not return here. This thread should continue servicing DDL operations. - } - - try{ - abortDDL(transactionState); - } - catch(Exception e){ - LOG.error("FATAL Exception calling abortDDL for transaction: " + transactionState.getTransactionId() + "Exception: " + e); - throw new UnsuccessfulDDLException(e); - } + loopExit = true; + } + catch (InterruptedException ie) {} + catch (CommitUnsuccessfulException cue) { + loopExit = true; + LOG.error("Exception at the time of aborting DDL transaction", cue); + savedCue = cue; + } + } while (loopExit == false); + abortDDL(transactionState); + if (savedCue != null) + throw savedCue; } if(LOG.isTraceEnabled()) LOG.trace("Abort -- EXIT txID: " + transactionState.getTransactionId()); } - void abortDDL(final TransactionState transactionState) throws UnsuccessfulDDLException + void abortDDL(final TransactionState transactionState) throws IOException, UnsuccessfulDDLException { //if tables were created, then they need to be dropped. ArrayList<String> createList = new ArrayList<String>(); @@ -2502,7 +2463,7 @@ public class TransactionManager { tmDDL.getRow(transactionState.getTransactionId(), state, createList, dropList, truncateList); retry = false; } - catch(Exception e){ + catch (IOException e){ LOG.error("Fatal Exception in abortDDL, Step: getRow. txID: " + transactionState.getTransactionId() + "Exception: " + e); if(retryCount == RETRY_ATTEMPTS) @@ -2524,11 +2485,11 @@ public class TransactionManager { retrySleep += TM_SLEEP_INCR; } } - }while (retryCount < RETRY_ATTEMPTS && retry == true); + } while (retryCount < RETRY_ATTEMPTS && retry == true); // if tables were recorded to be truncated on an upsert using load, // then they will be truncated on an abort transaction - if(state.toString().equals("VALID") && truncateList.size() > 0) + if (state.toString().equals("VALID") && truncateList.size() > 0) { if(LOG.isTraceEnabled()) LOG.trace("truncateList -- ENTRY txID: " + transactionState.getTransactionId()); @@ -2545,7 +2506,7 @@ public class TransactionManager { truncateTable(transactionState, tblName); retry = false; } - catch(Exception e){ + catch (IOException e){ LOG.error("Fatal exception in abortDDL, Step : truncateTable: TxID:" + transactionState.getTransactionId() + "Exception: " + e); if(retryCount == RETRY_ATTEMPTS) @@ -2594,7 +2555,7 @@ public class TransactionManager { if(LOG.isTraceEnabled()) LOG.trace(" TableNotFoundException exception in abortDDL deleteTable, Continuing: txID: " + transactionState.getTransactionId()); retry = false; } - catch(Exception e){ + catch (IOException e){ LOG.error("Fatal exception in abortDDL, Step : DeleteTable: TxID:" + transactionState.getTransactionId() + "Exception: " + e); if(retryCount == RETRY_ATTEMPTS) @@ -2634,42 +2595,7 @@ public class TransactionManager { retrySleep = TM_SLEEP; retry = true; String tblName = di.next(); - do - { - try { - enableTable(transactionState, tblName); - retry = false; - } - catch(TableNotDisabledException t){ - //Check for TableNotDisabledException, if that is the case, no further - //processing needed. This is not an error. Possible we are retrying the entire set of DDL changes - //because this transaction is being redriven for some reason. - if(LOG.isTraceEnabled()) LOG.trace(" TableNotDisabledException exception in abortDDL enableTable, Continuing: txID: " + transactionState.getTransactionId()); - retry = false; - } - catch(Exception e){ - LOG.error("Fatal exception in abortDDL, Step : enableTable: TxID:" + transactionState.getTransactionId() + "Exception: " + e); - if(retryCount == RETRY_ATTEMPTS) - { - LOG.error("Fatal Exception in doCommitDDL, Step: DeleteTable. Raising UnsuccessfulDDLException TxID:" + transactionState.getTransactionId() ); - - //Throw this exception after all retry attempts. - //Throwing a new exception gets out of the loop. - throw new UnsuccessfulDDLException(e); - } - - retryCount++; - if (retryCount < RETRY_ATTEMPTS) - { - try { - Thread.sleep(retrySleep); - } catch(InterruptedException ex) { - Thread.currentThread().interrupt(); - } - retrySleep += TM_SLEEP_INCR; - } - } - }while(retryCount < RETRY_ATTEMPTS && retry == true); + enableTable(transactionState, tblName); }//while } @@ -2683,14 +2609,13 @@ public class TransactionManager { tmDDL.deleteRow(transactionState.getTransactionId()); retry = false; } - catch(Exception e) + catch (IOException e) { LOG.error("Fatal Exception in abortDDL, Step: deleteRow. txID: " + transactionState.getTransactionId() + "Exception: " + e); if(retryCount == RETRY_ATTEMPTS) { LOG.error("Fatal Exception in abortDDL, Step: deleteRow. Raising UnsuccessfulDDLException. txID: " + transactionState.getTransactionId()); - //Throw this exception after all retry attempts. //Throwing a new exception gets out of the loop. throw new UnsuccessfulDDLException(e); @@ -2734,20 +2659,22 @@ public class TransactionManager { public void createTable(final TransactionState transactionState, HTableDescriptor desc, Object[] beginEndKeys) throws IOException { if (LOG.isTraceEnabled()) LOG.trace("createTable ENTRY, transactionState: " + transactionState.getTransactionId()); - + Admin admin = connection.getAdmin(); + try { if (beginEndKeys != null && beginEndKeys.length > 0) { byte[][] keys = new byte[beginEndKeys.length][]; for (int i = 0; i < beginEndKeys.length; i++){ keys[i] = (byte[])beginEndKeys[i]; if (LOG.isTraceEnabled()) LOG.trace("createTable with key #" + i + "value" + keys[i] + ") called."); } - hbadmin.createTable(desc, keys); + admin.createTable(desc, keys); } else { - hbadmin.createTable(desc); + admin.createTable(desc); } - // hbadmin.close(); - + } finally { + admin.close(); + } // Set transaction state object as participating in ddl transaction transactionState.setDDLTx(true); @@ -2979,7 +2906,7 @@ public class TransactionManager { return returnStatus; } - private void waitForCompletion(String tblName,HBaseAdmin admin) + private void waitForCompletion(String tblName,Admin admin) throws IOException { // poll for completion of an asynchronous operation boolean keepPolling = true; @@ -3004,8 +2931,10 @@ public class TransactionManager { public void alterTable(final TransactionState transactionState, String tblName, Object[] tableOptions) throws IOException { if (LOG.isTraceEnabled()) LOG.trace("alterTable ENTRY, transactionState: " + transactionState.getTransactionId()); - - HTableDescriptor htblDesc = hbadmin.getTableDescriptor(tblName.getBytes()); + Admin admin = connection.getAdmin(); + try { + TableName tableName = TableName.valueOf(tblName); + HTableDescriptor htblDesc = admin.getTableDescriptor(tableName); HColumnDescriptor[] families = htblDesc.getColumnFamilies(); HColumnDescriptor colDesc = families[0]; // Trafodion keeps SQL columns only in first column family int defaultVersionsValue = colDesc.getMaxVersions(); @@ -3014,14 +2943,16 @@ public class TransactionManager { setDescriptors(tableOptions,htblDesc /*out*/,colDesc /*out*/, defaultVersionsValue); if (status.tableDescriptorChanged()) { - hbadmin.modifyTable(tblName,htblDesc); - waitForCompletion(tblName,hbadmin); + admin.modifyTable(tableName,htblDesc); + waitForCompletion(tblName,admin); } else if (status.columnDescriptorChanged()) { - hbadmin.modifyColumn(tblName,colDesc); - waitForCompletion(tblName,hbadmin); + admin.modifyColumn(tableName,colDesc); + waitForCompletion(tblName,admin); } - hbadmin.close(); + } finally { + admin.close(); + } // Set transaction state object as participating in ddl transaction transactionState.setDDLTx(true); @@ -3068,24 +2999,23 @@ public class TransactionManager { //Called only by Abort or Commit processing. public void deleteTable(final TransactionState transactionState, final String tblName) - throws IOException{ + throws IOException { if (LOG.isTraceEnabled()) LOG.trace("deleteTable ENTRY, TxId: " + transactionState.getTransactionId() + " tableName " + tblName); - try{ - disableTable(transactionState, tblName); - } - catch (TableNotEnabledException e) { - //If table is not enabled, no need to throw exception. Continue. - //if (LOG.isTraceEnabled()) LOG.trace("deleteTable , TableNotEnabledException. This is a expected exception. Step: disableTable, TxId: " + - // transactionState.getTransactionId() + " TableName " + tblName + "Exception: " + e); - } - hbadmin.deleteTable(tblName); + disableTable(transactionState, tblName); + Admin admin = connection.getAdmin(); + admin.deleteTable(TableName.valueOf(tblName)); + admin.close(); } //Called only by Abort processing. public void enableTable(final TransactionState transactionState, String tblName) throws IOException{ if (LOG.isTraceEnabled()) LOG.trace("enableTable ENTRY, TxID: " + transactionState.getTransactionId() + " tableName " + tblName); - hbadmin.enableTable(tblName); + Admin admin = connection.getAdmin(); + TableName tableName = TableName.valueOf(tblName); + if (admin.isTableDisabled(tableName)) + admin.enableTable(TableName.valueOf(tblName)); + admin.close(); } // Called only by Abort processing to delete data from a table @@ -3093,23 +3023,27 @@ public class TransactionManager { throws IOException{ if (LOG.isTraceEnabled()) LOG.trace("truncateTable ENTRY, TxID: " + transactionState.getTransactionId() + "table: " + tblName); - - TableName tablename = TableName.valueOf(tblName); - HTableDescriptor hdesc = hbadmin.getTableDescriptor(tablename); + Admin admin = connection.getAdmin(); + TableName tableName = TableName.valueOf(tblName); + HTableDescriptor hdesc = admin.getTableDescriptor(tableName); // To be changed in 2.0 for truncate table - //hbadmin.truncateTable(tablename, true); - hbadmin.disableTable(tblName); - hbadmin.deleteTable(tblName); - hbadmin.createTable(hdesc); - hbadmin.close(); + if (admin.isTableEnabled(tableName)) + admin.disableTable(tableName); + admin.deleteTable(tableName); + admin.createTable(hdesc); + admin.close(); } //Called only by DoPrepare. public void disableTable(final TransactionState transactionState, String tblName) throws IOException{ if (LOG.isTraceEnabled()) LOG.trace("disableTable ENTRY, TxID: " + transactionState.getTransactionId() + " tableName " + tblName); - hbadmin.disableTable(tblName); + Admin admin = connection.getAdmin(); + TableName tableName = TableName.valueOf(tblName); + if (admin.isTableEnabled(tableName)) + admin.disableTable(tableName); + admin.close(); if (LOG.isTraceEnabled()) LOG.trace("disableTable EXIT, TxID: " + transactionState.getTransactionId() + " tableName " + tblName); } @@ -3125,35 +3059,7 @@ public class TransactionManager { HRegionInfo regionInfo = null; HTable table = null; - /* - * hostname and port no longer needed for RPC - final byte [] delimiter = ",".getBytes(); - String[] result = hostnamePort.split(new String(delimiter), 3); - - if (result.length < 2) - throw new IllegalArgumentException("Region array format is incorrect"); - - String hostname = result[0]; - int port = Integer.parseInt(result[1]); - LOG.debug("recoveryRequest regionInfo -- hostname:" + hostname + " port:" + port); - */ - - /* - * New way of parsing HRegionInfo used instead - ByteArrayInputStream lv_bis = new ByteArrayInputStream(regionArray); - DataInputStream lv_dis = new DataInputStream(lv_bis); - try { - regionInfo.readFields(lv_dis); - } catch (Exception e) { - StringWriter sw = new StringWriter(); - PrintWriter pw = new PrintWriter(sw); - e.printStackTrace(pw); - LOG.error("recoveryRequest exception in regionInfo.readFields() " + sw.toString()); - throw new Exception(); - } - */ - - regionInfo = HRegionInfo.parseFrom(regionArray); + regionInfo = HRegionInfo.parseFrom(regionArray); final String regionName = regionInfo.getRegionNameAsString(); final int tmID = tmid; http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/7fe96649/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalAggregationClient.java ---------------------------------------------------------------------- diff --git a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalAggregationClient.java b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalAggregationClient.java index 4c4bb24..8d54246 100644 --- a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalAggregationClient.java +++ b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalAggregationClient.java @@ -41,6 +41,7 @@ import org.apache.hadoop.hbase.Cell; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.HRegionInfo; import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.Connection; import org.apache.hadoop.hbase.client.transactional.TransactionalTable; import org.apache.hadoop.hbase.client.Result; import org.apache.hadoop.hbase.client.ResultScanner; @@ -81,13 +82,15 @@ public class TransactionalAggregationClient { private static final Log log = LogFactory.getLog(TransactionalAggregationClient.class); Configuration conf; + Connection connection; /** * Constructor with Conf object * @param cfg */ - public TransactionalAggregationClient(Configuration cfg) { + public TransactionalAggregationClient(Configuration cfg, Connection conn) { this.conf = cfg; + this.connection = conn; } /** @@ -108,7 +111,7 @@ public class TransactionalAggregationClient { throws Throwable { TransactionalTable table = null; try { - table = new TransactionalTable(tableName.getName()); + table = new TransactionalTable(tableName.getName(), connection); return max(transactionId, table, ci, scan); } finally { if (table != null) { @@ -207,7 +210,7 @@ public class TransactionalAggregationClient { throws Throwable { TransactionalTable table = null; try { - table = new TransactionalTable(tableName.getName()); + table = new TransactionalTable(tableName.getName(), connection); return min(transactionId, table, ci, scan); } finally { if (table != null) { @@ -294,7 +297,7 @@ public class TransactionalAggregationClient { throws Throwable { TransactionalTable table = null; try { - table = new TransactionalTable(tableName.getName()); + table = new TransactionalTable(tableName.getName(), connection); return rowCount(transactionId, table, ci, scan); } finally { if (table != null) { @@ -372,7 +375,7 @@ public class TransactionalAggregationClient { throws Throwable { TransactionalTable table = null; try { - table = new TransactionalTable(tableName.getName()); + table = new TransactionalTable(tableName.getName(), connection); return sum(transactionId, table, ci, scan); } finally { if (table != null) { @@ -449,7 +452,7 @@ public class TransactionalAggregationClient { throws Throwable { TransactionalTable table = null; try { - table = new TransactionalTable(tableName.getName()); + table = new TransactionalTable(tableName.getName(), connection); return getAvgArgs(transactionId, table, ci, scan); } finally { if (table != null) { @@ -648,7 +651,7 @@ public class TransactionalAggregationClient { Scan scan) throws Throwable { TransactionalTable table = null; try { - table = new TransactionalTable(tableName.getName()); + table = new TransactionalTable(tableName.getName(), connection); return std(transactionId, table, ci, scan); } finally { if (table != null) { @@ -765,7 +768,7 @@ public class TransactionalAggregationClient { Scan scan) throws Throwable { TransactionalTable table = null; try { - table = new TransactionalTable(tableName.getName()); + table = new TransactionalTable(tableName.getName(), connection); return median(transactionId, table, ci, scan); } finally { if (table != null) { http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/7fe96649/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalTable.java ---------------------------------------------------------------------- diff --git a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalTable.java b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalTable.java index f11b2f6..5d11e0f 100644 --- a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalTable.java +++ b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalTable.java @@ -47,14 +47,12 @@ import org.apache.commons.logging.LogFactory; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.Cell; -import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HRegionLocation; import org.apache.hadoop.hbase.KeyValue; import org.apache.hadoop.hbase.KeyValueUtil; import org.apache.hadoop.hbase.client.Delete; import org.apache.hadoop.hbase.client.Get; -import org.apache.hadoop.hbase.client.HConnectionManager; -import org.apache.hadoop.hbase.client.HConnection; +import org.apache.hadoop.hbase.client.Connection; import org.apache.hadoop.hbase.client.HTable; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.client.Result; @@ -99,38 +97,27 @@ import com.google.protobuf.ServiceException; */ public class TransactionalTable extends HTable implements TransactionalTableClient { static final Log LOG = LogFactory.getLog(RMInterface.class); - static private HConnection connection = null; - static Configuration config = HBaseConfiguration.create(); + static private Connection connection = null; static ExecutorService threadPool; static int retries = 15; static int delay = 1000; private String retryErrMsg = "Coprocessor result is null, retries exhausted"; - static { - config.set("hbase.hregion.impl", "org.apache.hadoop.hbase.regionserver.transactional.TransactionalRegion"); - try { - connection = HConnectionManager.createConnection(config); - } - catch (IOException ioe) { - LOG.error("Exception on TransactionTable anonymous static method. IOException while creating HConnection"); - } - threadPool = Executors.newCachedThreadPool(); - } - /** * @param tableName * @throws IOException */ - public TransactionalTable(final String tableName) throws IOException { - this(Bytes.toBytes(tableName)); + public TransactionalTable(final String tableName, Connection conn) throws IOException { + this(Bytes.toBytes(tableName), conn); } /** * @param tableName * @throws IOException */ - public TransactionalTable(final byte[] tableName) throws IOException { - super(tableName, connection, threadPool); + public TransactionalTable(final byte[] tableName, Connection conn) throws IOException { + super(tableName, conn, threadPool); + this.connection = conn; } private void addLocation(final TransactionState transactionState, HRegionLocation location) { @@ -733,9 +720,9 @@ public class TransactionalTable extends HTable implements TransactionalTableClie } } + private int maxKeyValueSize; - private int maxKeyValueSize; -public HRegionLocation getRegionLocation(byte[] row, boolean f) + public HRegionLocation getRegionLocation(byte[] row, boolean f) throws IOException { return super.getRegionLocation(row,f); } @@ -752,10 +739,6 @@ public HRegionLocation getRegionLocation(byte[] row, boolean f) throws IOException { super.flushCommits(); } - public HConnection getConnection() - { - return super.getConnection(); - } public byte[][] getEndKeys() throws IOException { http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/7fe96649/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalTableClient.java ---------------------------------------------------------------------- diff --git a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalTableClient.java b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalTableClient.java index 76e61d6..7453264 100644 --- a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalTableClient.java +++ b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalTableClient.java @@ -35,7 +35,7 @@ import org.apache.hadoop.hbase.client.Result; import org.apache.hadoop.hbase.client.ResultScanner; import org.apache.hadoop.hbase.client.Scan; import org.apache.hadoop.hbase.HRegionLocation; -import org.apache.hadoop.hbase.client.HConnection; +import org.apache.hadoop.hbase.client.Connection; import java.io.InterruptedIOException; import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException; @@ -123,7 +123,6 @@ public interface TransactionalTableClient { void setAutoFlush(boolean autoFlush, boolean b); org.apache.hadoop.conf.Configuration getConfiguration(); void flushCommits() throws IOException; - HConnection getConnection(); byte[][] getEndKeys() throws IOException; http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/7fe96649/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/UnsuccessfulDDLException.java ---------------------------------------------------------------------- diff --git a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/UnsuccessfulDDLException.java b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/UnsuccessfulDDLException.java index 9e7baf5..95e58d7 100644 --- a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/UnsuccessfulDDLException.java +++ b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/UnsuccessfulDDLException.java @@ -22,11 +22,12 @@ **/ package org.apache.hadoop.hbase.client.transactional; +import java.io.IOException; /** Thrown when a transaction specific to DDL operation cannot be committed. * */ -public class UnsuccessfulDDLException extends Exception { +public class UnsuccessfulDDLException extends IOException { private static final long serialVersionUID = 7062921444531109202L; http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/7fe96649/core/sqf/src/seatrans/tm/hbasetmlib2/src/main/java/org/trafodion/dtm/HBaseAuditControlPoint.java ---------------------------------------------------------------------- diff --git a/core/sqf/src/seatrans/tm/hbasetmlib2/src/main/java/org/trafodion/dtm/HBaseAuditControlPoint.java b/core/sqf/src/seatrans/tm/hbasetmlib2/src/main/java/org/trafodion/dtm/HBaseAuditControlPoint.java index 12128f5..ba81304 100644 --- a/core/sqf/src/seatrans/tm/hbasetmlib2/src/main/java/org/trafodion/dtm/HBaseAuditControlPoint.java +++ b/core/sqf/src/seatrans/tm/hbasetmlib2/src/main/java/org/trafodion/dtm/HBaseAuditControlPoint.java @@ -36,9 +36,10 @@ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.Cell; import org.apache.hadoop.hbase.CellUtil; -import org.apache.hadoop.hbase.client.HBaseAdmin; -import org.apache.hadoop.hbase.client.HConnection; -import org.apache.hadoop.hbase.client.HConnectionManager; +import org.apache.hadoop.hbase.client.Table; +import org.apache.hadoop.hbase.client.Admin; +import org.apache.hadoop.hbase.client.Connection; +import org.apache.hadoop.hbase.client.ConnectionFactory; import org.apache.hadoop.hbase.client.HTable; import org.apache.hadoop.hbase.client.HTableInterface; import org.apache.hadoop.hbase.client.Get; @@ -53,7 +54,6 @@ import org.apache.hadoop.hbase.client.transactional.CommitUnsuccessfulException; import org.apache.hadoop.hbase.client.transactional.UnknownTransactionException; import org.apache.hadoop.hbase.client.transactional.HBaseBackedTransactionLogger; import org.apache.hadoop.hbase.client.transactional.TransactionRegionLocation; -import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HColumnDescriptor; import org.apache.hadoop.hbase.HRegionInfo; import org.apache.hadoop.hbase.HRegionLocation; @@ -79,7 +79,7 @@ public class HBaseAuditControlPoint { static final Log LOG = LogFactory.getLog(HBaseAuditControlPoint.class); private static long currControlPt; - private static HBaseAdmin admin; + private Connection connection; private Configuration config; private static String CONTROL_POINT_TABLE_NAME; private static final byte[] CONTROL_POINT_FAMILY = Bytes.toBytes("cpf"); @@ -88,9 +88,10 @@ public class HBaseAuditControlPoint { private boolean useAutoFlush; private boolean disableBlockCache; - public HBaseAuditControlPoint(Configuration config) throws IOException { + public HBaseAuditControlPoint(Configuration config, Connection connection) throws IOException { if (LOG.isTraceEnabled()) LOG.trace("Enter HBaseAuditControlPoint constructor()"); this.config = config; + this.connection = connection; CONTROL_POINT_TABLE_NAME = config.get("CONTROL_POINT_TABLE_NAME"); HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(CONTROL_POINT_TABLE_NAME)); HColumnDescriptor hcol = new HColumnDescriptor(CONTROL_POINT_FAMILY); @@ -107,7 +108,6 @@ public class HBaseAuditControlPoint { } desc.addFamily(hcol); - admin = new HBaseAdmin(config); useAutoFlush = true; String autoFlush = System.getenv("TM_TLOG_AUTO_FLUSH"); @@ -116,10 +116,11 @@ public class HBaseAuditControlPoint { if (LOG.isDebugEnabled()) LOG.debug("autoFlush != null"); } LOG.info("useAutoFlush is " + useAutoFlush); - - boolean lvControlPointExists = admin.tableExists(CONTROL_POINT_TABLE_NAME); + Admin admin = connection.getAdmin(); + boolean lvControlPointExists = admin.tableExists(TableName.valueOf(CONTROL_POINT_TABLE_NAME)); if (LOG.isDebugEnabled()) LOG.debug("HBaseAuditControlPoint lvControlPointExists " + lvControlPointExists); currControlPt = -1; + admin.close(); if (lvControlPointExists == false) { try { if (LOG.isDebugEnabled()) LOG.debug("Creating the table " + CONTROL_POINT_TABLE_NAME); @@ -253,9 +254,9 @@ public class HBaseAuditControlPoint { if (LOG.isTraceEnabled()) LOG.trace("getNextAuditSeqNum for node: " + nid); // We need to open the appropriate control point table and read the value from it - HTableInterface remoteTable; + Table remoteTable; String lv_tName = new String("TRAFODION._DTM_.TLOG" + String.valueOf(nid) + "_CONTROL_POINT"); - HConnection remoteConnection = HConnectionManager.createConnection(this.config); + Connection remoteConnection = ConnectionFactory.createConnection(this.config); remoteTable = remoteConnection.getTable(TableName.valueOf(lv_tName)); long highValue = -1; http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/7fe96649/core/sqf/src/seatrans/tm/hbasetmlib2/src/main/java/org/trafodion/dtm/HBaseTxClient.java ---------------------------------------------------------------------- diff --git a/core/sqf/src/seatrans/tm/hbasetmlib2/src/main/java/org/trafodion/dtm/HBaseTxClient.java b/core/sqf/src/seatrans/tm/hbasetmlib2/src/main/java/org/trafodion/dtm/HBaseTxClient.java index b06d6c4..e12e365 100644 --- a/core/sqf/src/seatrans/tm/hbasetmlib2/src/main/java/org/trafodion/dtm/HBaseTxClient.java +++ b/core/sqf/src/seatrans/tm/hbasetmlib2/src/main/java/org/trafodion/dtm/HBaseTxClient.java @@ -41,6 +41,8 @@ import org.apache.commons.logging.LogFactory; import org.apache.commons.codec.binary.Hex; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.client.Connection; +import org.apache.hadoop.hbase.client.ConnectionFactory; import org.apache.hadoop.hbase.client.HTable; import org.apache.hadoop.hbase.client.transactional.TransactionManager; import org.apache.hadoop.hbase.client.transactional.TransactionState; @@ -81,6 +83,7 @@ import java.util.concurrent.ConcurrentHashMap; public class HBaseTxClient { static final Log LOG = LogFactory.getLog(HBaseTxClient.class); + private static Connection connection; private static TmAuditTlog tLog; private static HBaseTmZK tmZK; private static RecoveryThread recovThread; @@ -120,8 +123,10 @@ public class HBaseTxClient { setupLog4j(); if (LOG.isDebugEnabled()) LOG.debug("Enter init, hBasePath:" + hBasePath); if (LOG.isTraceEnabled()) LOG.trace("mapTransactionStates " + mapTransactionStates + " entries " + mapTransactionStates.size()); - config = HBaseConfiguration.create(); - + if (config == null) { + config = HBaseConfiguration.create(); + connection = ConnectionFactory.createConnection(config); + } config.set("hbase.zookeeper.quorum", zkServers); config.set("hbase.zookeeper.property.clientPort",zkPort); config.set("hbase.rootdir", hBasePath); @@ -168,10 +173,10 @@ public class HBaseTxClient { } if (useTlog) { - tLog = new TmAuditTlog(config); + tLog = new TmAuditTlog(config, connection); } try { - trxManager = TransactionManager.getInstance(config); + trxManager = TransactionManager.getInstance(config, connection); } catch (IOException e ){ LOG.error("Unable to create TransactionManager, throwing exception", e); throw e; @@ -197,7 +202,10 @@ public class HBaseTxClient { setupLog4j(); if (LOG.isDebugEnabled()) LOG.debug("Enter init(" + dtmid + ")"); - config = HBaseConfiguration.create(); + if (config == null) { + config = HBaseConfiguration.create(); + connection = ConnectionFactory.createConnection(config); + } config.set("hbase.hregion.impl", "org.apache.hadoop.hbase.regionserver.transactional.TransactionalRegion"); config.set("hbase.hlog.splitter.impl", "org.apache.hadoop.hbase.regionserver.transactional.THLogSplitter"); config.set("dtmid", String.valueOf(dtmid)); @@ -232,7 +240,7 @@ public class HBaseTxClient { } if (useDDLTrans) - tmDDL = new TmDDL(config); + tmDDL = new TmDDL(config, connection); useForgotten = true; String useAuditRecords = System.getenv("TM_ENABLE_FORGOTTEN_RECORDS"); @@ -270,9 +278,9 @@ public class HBaseTxClient { LOG.error("TM_ENABLE_TLOG_WRITES is not valid in ms.env"); } if (useTlog) { - tLog = new TmAuditTlog(config); + tLog = new TmAuditTlog(config, connection); } - trxManager = TransactionManager.getInstance(config); + trxManager = TransactionManager.getInstance(config, connection); if(useDDLTrans) trxManager.init(tmDDL); @@ -420,14 +428,7 @@ public class HBaseTxClient { try { trxManager.abort(ts); - } catch(IOException e) { - synchronized(mapLock) { - mapTransactionStates.remove(transactionID); - } - LOG.error("Returning from HBaseTxClient:abortTransaction, txid: " + transactionID + " retval: EXCEPTION", e); - return TransReturnCode.RET_EXCEPTION.getShort(); - } - catch (UnsuccessfulDDLException ddle) { + } catch (UnsuccessfulDDLException ddle) { LOG.error("FATAL DDL Exception from HBaseTxClient:abort, WAITING INDEFINETLY !! retval: " + TransReturnCode.RET_EXCEPTION.toString() + " UnsuccessfulDDLException" + " txid: " + transactionID, ddle); //Reaching here means several attempts to perform the DDL operation has failed in abort phase. @@ -452,7 +453,15 @@ public class HBaseTxClient { } while (loopBack); } return TransReturnCode.RET_EXCEPTION.getShort(); + } + catch(IOException e) { + synchronized(mapLock) { + mapTransactionStates.remove(transactionID); + } + LOG.error("Returning from HBaseTxClient:abortTransaction, txid: " + transactionID + " retval: EXCEPTION", e); + return TransReturnCode.RET_EXCEPTION.getShort(); } + if (useTlog && useForgotten) { if (forceForgotten) { tLog.putSingleRecord(transactionID, -1, "FORGOTTEN", null, true);
