You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@omid.apache.org by ik...@apache.org on 2016/04/21 02:02:57 UTC
[40/52] [abbrv] incubator-omid git commit: Move com.yahoo ->
org.apache
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/hbase-client/src/main/java/com/yahoo/omid/transaction/TTable.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/com/yahoo/omid/transaction/TTable.java b/hbase-client/src/main/java/com/yahoo/omid/transaction/TTable.java
deleted file mode 100644
index 53b62eb..0000000
--- a/hbase-client/src/main/java/com/yahoo/omid/transaction/TTable.java
+++ /dev/null
@@ -1,770 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package com.yahoo.omid.transaction;
-
-import com.google.common.base.Function;
-import com.google.common.base.Optional;
-import com.google.common.base.Predicate;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Multimaps;
-import com.yahoo.omid.committable.CommitTable.CommitTimestamp;
-import com.yahoo.omid.transaction.HBaseTransactionManager.CommitTimestampLocatorImpl;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.CellUtil;
-import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.hadoop.hbase.HConstants;
-import org.apache.hadoop.hbase.HTableDescriptor;
-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.HTable;
-import org.apache.hadoop.hbase.client.HTableInterface;
-import org.apache.hadoop.hbase.client.Mutation;
-import org.apache.hadoop.hbase.client.Put;
-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.io.TimeRange;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.Closeable;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.NavigableMap;
-import java.util.NavigableSet;
-
-/**
- * Provides transactional methods for accessing and modifying a given snapshot
- * of data identified by an opaque {@link Transaction} object. It mimics the
- * behavior in {@link org.apache.hadoop.hbase.client.HTableInterface}
- */
-public class TTable implements Closeable {
-
- private static Logger LOG = LoggerFactory.getLogger(TTable.class);
-
- private final HTableInterface healerTable;
-
- private HTableInterface table;
-
- // ----------------------------------------------------------------------------------------------------------------
- // Construction
- // ----------------------------------------------------------------------------------------------------------------
-
- public TTable(Configuration conf, byte[] tableName) throws IOException {
- this(new HTable(conf, tableName));
- }
-
- public TTable(String tableName) throws IOException {
- this(HBaseConfiguration.create(), Bytes.toBytes(tableName));
- }
-
- public TTable(Configuration conf, String tableName) throws IOException {
- this(conf, Bytes.toBytes(tableName));
- }
-
- public TTable(HTableInterface hTable) throws IOException {
- table = hTable;
- healerTable = new HTable(table.getConfiguration(), table.getTableName());
- }
-
- public TTable(HTableInterface hTable, HTableInterface healerTable) throws IOException {
- table = hTable;
- this.healerTable = healerTable;
- }
-
- // ----------------------------------------------------------------------------------------------------------------
- // Closeable implementation
- // ----------------------------------------------------------------------------------------------------------------
-
- /**
- * Releases any resources held or pending changes in internal buffers.
- *
- * @throws IOException
- * if a remote or network exception occurs.
- */
- @Override
- public void close() throws IOException {
- table.close();
- healerTable.close();
- }
-
- // ----------------------------------------------------------------------------------------------------------------
- // Transactional operations
- // ----------------------------------------------------------------------------------------------------------------
-
- /**
- * Transactional version of {@link HTableInterface#get(Get get)}
- */
- public Result get(Transaction tx, final Get get) throws IOException {
-
- throwExceptionIfOpSetsTimerange(get);
-
- HBaseTransaction transaction = enforceHBaseTransactionAsParam(tx);
-
- final long readTimestamp = transaction.getStartTimestamp();
- final Get tsget = new Get(get.getRow()).setFilter(get.getFilter());
- TimeRange timeRange = get.getTimeRange();
- long startTime = timeRange.getMin();
- long endTime = Math.min(timeRange.getMax(), readTimestamp + 1);
- tsget.setTimeRange(startTime, endTime).setMaxVersions(1);
- Map<byte[], NavigableSet<byte[]>> kvs = get.getFamilyMap();
- for (Map.Entry<byte[], NavigableSet<byte[]>> entry : kvs.entrySet()) {
- byte[] family = entry.getKey();
- NavigableSet<byte[]> qualifiers = entry.getValue();
- if (qualifiers == null || qualifiers.isEmpty()) {
- tsget.addFamily(family);
- } else {
- for (byte[] qualifier : qualifiers) {
- tsget.addColumn(family, qualifier);
- tsget.addColumn(family, CellUtils.addShadowCellSuffix(qualifier));
- }
- }
- }
- LOG.trace("Initial Get = {}", tsget);
-
- // Return the KVs that belong to the transaction snapshot, ask for more
- // versions if needed
- Result result = table.get(tsget);
- List<Cell> filteredKeyValues = Collections.emptyList();
- if (!result.isEmpty()) {
- filteredKeyValues = filterCellsForSnapshot(result.listCells(), transaction, tsget.getMaxVersions());
- }
-
- return Result.create(filteredKeyValues);
- }
-
- /**
- * Transactional version of {@link HTableInterface#delete(Delete delete)}
- */
- public void delete(Transaction tx, Delete delete) throws IOException {
-
- throwExceptionIfOpSetsTimerange(delete);
-
- HBaseTransaction transaction = enforceHBaseTransactionAsParam(tx);
-
- final long startTimestamp = transaction.getStartTimestamp();
- boolean issueGet = false;
-
- final Put deleteP = new Put(delete.getRow(), startTimestamp);
- final Get deleteG = new Get(delete.getRow());
- Map<byte[], List<Cell>> fmap = delete.getFamilyCellMap();
- if (fmap.isEmpty()) {
- issueGet = true;
- }
- for (List<Cell> cells : fmap.values()) {
- for (Cell cell : cells) {
- CellUtils.validateCell(cell, startTimestamp);
- switch (KeyValue.Type.codeToType(cell.getTypeByte())) {
- case DeleteColumn:
- deleteP.add(CellUtil.cloneFamily(cell),
- CellUtil.cloneQualifier(cell),
- startTimestamp,
- CellUtils.DELETE_TOMBSTONE);
- transaction.addWriteSetElement(
- new HBaseCellId(table,
- delete.getRow(),
- CellUtil.cloneFamily(cell),
- CellUtil.cloneQualifier(cell),
- cell.getTimestamp()));
- break;
- case DeleteFamily:
- deleteG.addFamily(CellUtil.cloneFamily(cell));
- issueGet = true;
- break;
- case Delete:
- if (cell.getTimestamp() == HConstants.LATEST_TIMESTAMP) {
- deleteP.add(CellUtil.cloneFamily(cell),
- CellUtil.cloneQualifier(cell),
- startTimestamp,
- CellUtils.DELETE_TOMBSTONE);
- transaction.addWriteSetElement(
- new HBaseCellId(table,
- delete.getRow(),
- CellUtil.cloneFamily(cell),
- CellUtil.cloneQualifier(cell),
- cell.getTimestamp()));
- break;
- } else {
- throw new UnsupportedOperationException(
- "Cannot delete specific versions on Snapshot Isolation.");
- }
- default:
- break;
- }
- }
- }
- if (issueGet) {
- // It's better to perform a transactional get to avoid deleting more
- // than necessary
- Result result = this.get(transaction, deleteG);
- if (!result.isEmpty()) {
- for (Entry<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> entryF : result.getMap()
- .entrySet()) {
- byte[] family = entryF.getKey();
- for (Entry<byte[], NavigableMap<Long, byte[]>> entryQ : entryF.getValue().entrySet()) {
- byte[] qualifier = entryQ.getKey();
- deleteP.add(family, qualifier, CellUtils.DELETE_TOMBSTONE);
- transaction.addWriteSetElement(new HBaseCellId(table, delete.getRow(), family, qualifier, transaction.getStartTimestamp()));
- }
- }
- }
- }
-
- if (!deleteP.isEmpty()) {
- table.put(deleteP);
- }
-
- }
-
- /**
- * Transactional version of {@link HTableInterface#put(Put put)}
- */
- public void put(Transaction tx, Put put) throws IOException {
-
- throwExceptionIfOpSetsTimerange(put);
-
- HBaseTransaction transaction = enforceHBaseTransactionAsParam(tx);
-
- final long startTimestamp = transaction.getStartTimestamp();
- // create put with correct ts
- final Put tsput = new Put(put.getRow(), startTimestamp);
- Map<byte[], List<Cell>> kvs = put.getFamilyCellMap();
- for (List<Cell> kvl : kvs.values()) {
- for (Cell c : kvl) {
- CellUtils.validateCell(c, startTimestamp);
- // Reach into keyvalue to update timestamp.
- // It's not nice to reach into keyvalue internals,
- // but we want to avoid having to copy the whole thing
- KeyValue kv = KeyValueUtil.ensureKeyValue(c);
- Bytes.putLong(kv.getValueArray(), kv.getTimestampOffset(), startTimestamp);
- tsput.add(kv);
-
- transaction.addWriteSetElement(
- new HBaseCellId(table,
- CellUtil.cloneRow(kv),
- CellUtil.cloneFamily(kv),
- CellUtil.cloneQualifier(kv),
- kv.getTimestamp()));
- }
- }
-
- table.put(tsput);
- }
-
- /**
- * Transactional version of {@link HTableInterface#getScanner(Scan scan)}
- */
- public ResultScanner getScanner(Transaction tx, Scan scan) throws IOException {
-
- throwExceptionIfOpSetsTimerange(scan);
-
- HBaseTransaction transaction = enforceHBaseTransactionAsParam(tx);
-
- Scan tsscan = new Scan(scan);
- tsscan.setMaxVersions(1);
- tsscan.setTimeRange(0, transaction.getStartTimestamp() + 1);
- Map<byte[], NavigableSet<byte[]>> kvs = scan.getFamilyMap();
- for (Map.Entry<byte[], NavigableSet<byte[]>> entry : kvs.entrySet()) {
- byte[] family = entry.getKey();
- NavigableSet<byte[]> qualifiers = entry.getValue();
- if (qualifiers == null) {
- continue;
- }
- for (byte[] qualifier : qualifiers) {
- tsscan.addColumn(family, CellUtils.addShadowCellSuffix(qualifier));
- }
- }
- return new TransactionalClientScanner(transaction, tsscan, 1);
- }
-
- /**
- * Filters the raw results returned from HBase and returns only those
- * belonging to the current snapshot, as defined by the transaction
- * object. If the raw results don't contain enough information for a
- * particular qualifier, it will request more versions from HBase.
- *
- * @param rawCells
- * Raw cells that we are going to filter
- * @param transaction
- * Defines the current snapshot
- * @param versionsToRequest
- * Number of versions requested from hbase
- * @return Filtered KVs belonging to the transaction snapshot
- * @throws IOException
- */
- List<Cell> filterCellsForSnapshot(List<Cell> rawCells, HBaseTransaction transaction,
- int versionsToRequest) throws IOException {
-
- assert (rawCells != null && transaction != null && versionsToRequest >= 1);
-
- List<Cell> keyValuesInSnapshot = new ArrayList<>();
- List<Get> pendingGetsList = new ArrayList<>();
-
- int numberOfVersionsToFetch = versionsToRequest * 2;
- if (numberOfVersionsToFetch < 1) {
- numberOfVersionsToFetch = versionsToRequest;
- }
-
- Map<Long, Long> commitCache = buildCommitCache(rawCells);
-
- for (Collection<Cell> columnCells : groupCellsByColumnFilteringShadowCells(rawCells)) {
- boolean snapshotValueFound = false;
- Cell oldestCell = null;
- for (Cell cell : columnCells) {
- if (isCellInSnapshot(cell, transaction, commitCache)) {
- if (!CellUtil.matchingValue(cell, CellUtils.DELETE_TOMBSTONE)) {
- keyValuesInSnapshot.add(cell);
- }
- snapshotValueFound = true;
- break;
- }
- oldestCell = cell;
- }
- if (!snapshotValueFound) {
- assert (oldestCell != null);
- Get pendingGet = createPendingGet(oldestCell, numberOfVersionsToFetch);
- pendingGetsList.add(pendingGet);
- }
- }
-
- if (!pendingGetsList.isEmpty()) {
- Result[] pendingGetsResults = table.get(pendingGetsList);
- for (Result pendingGetResult : pendingGetsResults) {
- if (!pendingGetResult.isEmpty()) {
- keyValuesInSnapshot.addAll(
- filterCellsForSnapshot(pendingGetResult.listCells(), transaction, numberOfVersionsToFetch));
- }
- }
- }
-
- Collections.sort(keyValuesInSnapshot, KeyValue.COMPARATOR);
-
- assert (keyValuesInSnapshot.size() <= rawCells.size());
- return keyValuesInSnapshot;
- }
-
- private Map<Long, Long> buildCommitCache(List<Cell> rawCells) {
-
- Map<Long, Long> commitCache = new HashMap<>();
-
- for (Cell cell : rawCells) {
- if (CellUtils.isShadowCell(cell)) {
- commitCache.put(cell.getTimestamp(), Bytes.toLong(CellUtil.cloneValue(cell)));
- }
- }
-
- return commitCache;
- }
-
- private boolean isCellInSnapshot(Cell kv, HBaseTransaction transaction, Map<Long, Long> commitCache)
- throws IOException {
-
- long startTimestamp = transaction.getStartTimestamp();
-
- if (kv.getTimestamp() == startTimestamp) {
- return true;
- }
-
- Optional<Long> commitTimestamp =
- tryToLocateCellCommitTimestamp(transaction.getTransactionManager(), transaction.getEpoch(), kv, commitCache);
-
- return commitTimestamp.isPresent() && commitTimestamp.get() < startTimestamp;
- }
-
- private Get createPendingGet(Cell cell, int versionCount) throws IOException {
-
- Get pendingGet = new Get(CellUtil.cloneRow(cell));
- pendingGet.addColumn(CellUtil.cloneFamily(cell), CellUtil.cloneQualifier(cell));
- pendingGet.addColumn(CellUtil.cloneFamily(cell), CellUtils.addShadowCellSuffix(cell.getQualifierArray(),
- cell.getQualifierOffset(),
- cell.getQualifierLength()));
- pendingGet.setMaxVersions(versionCount);
- pendingGet.setTimeRange(0, cell.getTimestamp());
-
- return pendingGet;
- }
-
- private Optional<Long> tryToLocateCellCommitTimestamp(AbstractTransactionManager transactionManager,
- long epoch,
- Cell cell,
- Map<Long, Long> commitCache)
- throws IOException {
-
- CommitTimestamp tentativeCommitTimestamp =
- transactionManager.locateCellCommitTimestamp(
- cell.getTimestamp(),
- epoch,
- new CommitTimestampLocatorImpl(
- new HBaseCellId(table,
- CellUtil.cloneRow(cell),
- CellUtil.cloneFamily(cell),
- CellUtil.cloneQualifier(cell),
- cell.getTimestamp()),
- commitCache));
-
- // If transaction that added the cell was invalidated
- if (!tentativeCommitTimestamp.isValid()) {
- return Optional.absent();
- }
-
- switch (tentativeCommitTimestamp.getLocation()) {
- case COMMIT_TABLE:
- // If the commit timestamp is found in the persisted commit table,
- // that means the writing process of the shadow cell in the post
- // commit phase of the client probably failed, so we heal the shadow
- // cell with the right commit timestamp for avoiding further reads to
- // hit the storage
- healShadowCell(cell, tentativeCommitTimestamp.getValue());
- case CACHE:
- case SHADOW_CELL:
- return Optional.of(tentativeCommitTimestamp.getValue());
- case NOT_PRESENT:
- return Optional.absent();
- default:
- assert (false);
- return Optional.absent();
- }
- }
-
- void healShadowCell(Cell cell, long commitTimestamp) {
- Put put = new Put(CellUtil.cloneRow(cell));
- byte[] family = CellUtil.cloneFamily(cell);
- byte[] shadowCellQualifier = CellUtils.addShadowCellSuffix(cell.getQualifierArray(),
- cell.getQualifierOffset(),
- cell.getQualifierLength());
- put.add(family, shadowCellQualifier, cell.getTimestamp(), Bytes.toBytes(commitTimestamp));
- try {
- healerTable.put(put);
- } catch (IOException e) {
- LOG.warn("Failed healing shadow cell for kv {}", cell, e);
- }
- }
-
- protected class TransactionalClientScanner implements ResultScanner {
- private HBaseTransaction state;
- private ResultScanner innerScanner;
- private int maxVersions;
-
- TransactionalClientScanner(HBaseTransaction state, Scan scan, int maxVersions)
- throws IOException {
- this.state = state;
- this.innerScanner = table.getScanner(scan);
- this.maxVersions = maxVersions;
- }
-
-
- @Override
- public Result next() throws IOException {
- List<Cell> filteredResult = Collections.emptyList();
- while (filteredResult.isEmpty()) {
- Result result = innerScanner.next();
- if (result == null) {
- return null;
- }
- if (!result.isEmpty()) {
- filteredResult = filterCellsForSnapshot(result.listCells(), state, maxVersions);
- }
- }
- return Result.create(filteredResult);
- }
-
- // In principle no need to override, copied from super.next(int) to make
- // sure it works even if super.next(int)
- // changes its implementation
- @Override
- public Result[] next(int nbRows) throws IOException {
- // Collect values to be returned here
- ArrayList<Result> resultSets = new ArrayList<>(nbRows);
- for (int i = 0; i < nbRows; i++) {
- Result next = next();
- if (next != null) {
- resultSets.add(next);
- } else {
- break;
- }
- }
- return resultSets.toArray(new Result[resultSets.size()]);
- }
-
- @Override
- public void close() {
- innerScanner.close();
- }
-
- @Override
- public Iterator<Result> iterator() {
- return new ResultIterator(this);
- }
-
- // ------------------------------------------------------------------------------------------------------------
- // --------------------------------- Helper class for TransactionalClientScanner ------------------------------
- // ------------------------------------------------------------------------------------------------------------
-
- class ResultIterator implements Iterator<Result> {
-
- TransactionalClientScanner scanner;
- Result currentResult;
-
- ResultIterator(TransactionalClientScanner scanner) {
- try {
- this.scanner = scanner;
- currentResult = scanner.next();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public boolean hasNext() {
- return currentResult != null && !currentResult.isEmpty();
- }
-
- @Override
- public Result next() {
- try {
- Result result = currentResult;
- currentResult = scanner.next();
- return result;
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public void remove() {
- throw new RuntimeException("Not implemented");
- }
-
- }
-
- }
-
- /**
- * Delegates to {@link HTable#getTableName()}
- */
- public byte[] getTableName() {
- return table.getTableName();
- }
-
- /**
- * Delegates to {@link HTable#getConfiguration()}
- */
- public Configuration getConfiguration() {
- return table.getConfiguration();
- }
-
- /**
- * Delegates to {@link HTable#getTableDescriptor()}
- */
- public HTableDescriptor getTableDescriptor() throws IOException {
- return table.getTableDescriptor();
- }
-
- /**
- * Transactional version of {@link HTableInterface#exists(Get get)}
- */
- public boolean exists(Transaction transaction, Get get) throws IOException {
- Result result = get(transaction, get);
- return !result.isEmpty();
- }
-
- /* TODO What should we do with this methods???
- * @Override public void batch(Transaction transaction, List<? extends Row>
- * actions, Object[] results) throws IOException, InterruptedException {}
- *
- * @Override public Object[] batch(Transaction transaction, List<? extends
- * Row> actions) throws IOException, InterruptedException {}
- *
- * @Override public <R> void batchCallback(Transaction transaction, List<?
- * extends Row> actions, Object[] results, Callback<R> callback) throws
- * IOException, InterruptedException {}
- *
- * @Override public <R> Object[] batchCallback(List<? extends Row> actions,
- * Callback<R> callback) throws IOException, InterruptedException {}
- */
-
- /**
- * Transactional version of {@link HTableInterface#get(List<Get> gets)}
- */
- public Result[] get(Transaction transaction, List<Get> gets) throws IOException {
- Result[] results = new Result[gets.size()];
- int i = 0;
- for (Get get : gets) {
- results[i++] = get(transaction, get);
- }
- return results;
- }
-
- /**
- * Transactional version of {@link HTableInterface#getScanner(byte[] family)}
- */
- public ResultScanner getScanner(Transaction transaction, byte[] family) throws IOException {
- Scan scan = new Scan();
- scan.addFamily(family);
- return getScanner(transaction, scan);
- }
-
- /**
- * Transactional version of {@link HTableInterface#getScanner(byte[] family, byte[] qualifier)}
- */
- public ResultScanner getScanner(Transaction transaction, byte[] family, byte[] qualifier)
- throws IOException {
- Scan scan = new Scan();
- scan.addColumn(family, qualifier);
- return getScanner(transaction, scan);
- }
-
- /**
- * Transactional version of {@link HTableInterface#put(List<Put> puts)}
- */
- public void put(Transaction transaction, List<Put> puts) throws IOException {
- for (Put put : puts) {
- put(transaction, put);
- }
- }
-
- /**
- * Transactional version of {@link HTableInterface#delete(List<Delete> deletes)}
- */
- public void delete(Transaction transaction, List<Delete> deletes) throws IOException {
- for (Delete delete : deletes) {
- delete(transaction, delete);
- }
- }
-
- /**
- * Provides access to the underliying HTable in order to configure it or to
- * perform unsafe (non-transactional) operations. The latter would break the
- * transactional guarantees of the whole system.
- *
- * @return The underlying HTable object
- */
- public HTableInterface getHTable() {
- return table;
- }
-
- /**
- * Delegates to {@link HTable#setAutoFlush(boolean autoFlush)}
- */
- public void setAutoFlush(boolean autoFlush) {
- table.setAutoFlush(autoFlush, true);
- }
-
- /**
- * Delegates to {@link HTable#isAutoFlush()}
- */
- public boolean isAutoFlush() {
- return table.isAutoFlush();
- }
-
- /**
- * Delegates to {@link HTable.getWriteBufferSize()}
- */
- public long getWriteBufferSize() {
- return table.getWriteBufferSize();
- }
-
- /**
- * Delegates to {@link HTable.setWriteBufferSize()}
- */
- public void setWriteBufferSize(long writeBufferSize) throws IOException {
- table.setWriteBufferSize(writeBufferSize);
- }
-
- /**
- * Delegates to {@link HTable.flushCommits()}
- */
- public void flushCommits() throws IOException {
- table.flushCommits();
- }
-
- // ----------------------------------------------------------------------------------------------------------------
- // Helper methods
- // ----------------------------------------------------------------------------------------------------------------
-
- private void throwExceptionIfOpSetsTimerange(Get getOperation) {
- TimeRange tr = getOperation.getTimeRange();
- checkTimerangeIsSetToDefaultValuesOrThrowException(tr);
- }
-
- private void throwExceptionIfOpSetsTimerange(Scan scanOperation) {
- TimeRange tr = scanOperation.getTimeRange();
- checkTimerangeIsSetToDefaultValuesOrThrowException(tr);
- }
-
- private void checkTimerangeIsSetToDefaultValuesOrThrowException(TimeRange tr) {
- if (tr.getMin() != 0L || tr.getMax() != Long.MAX_VALUE) {
- throw new IllegalArgumentException(
- "Timestamp/timerange not allowed in transactional user operations");
- }
- }
-
- private void throwExceptionIfOpSetsTimerange(Mutation userOperation) {
- if (userOperation.getTimeStamp() != HConstants.LATEST_TIMESTAMP) {
- throw new IllegalArgumentException(
- "Timestamp not allowed in transactional user operations");
- }
- }
-
- private HBaseTransaction enforceHBaseTransactionAsParam(Transaction tx) {
- if (tx instanceof HBaseTransaction) {
- return (HBaseTransaction) tx;
- } else {
- throw new IllegalArgumentException(
- String.format("The transaction object passed %s is not an instance of HBaseTransaction",
- tx.getClass().getName()));
- }
- }
-
- static ImmutableList<Collection<Cell>> groupCellsByColumnFilteringShadowCells(List<Cell> rawCells) {
-
- Predicate<Cell> shadowCellFilter = new Predicate<Cell>() {
-
- @Override
- public boolean apply(Cell cell) {
- return !CellUtils.isShadowCell(cell);
- }
-
- };
-
- Function<Cell, ColumnWrapper> cellToColumnWrapper = new Function<Cell, ColumnWrapper>() {
-
- @Override
- public ColumnWrapper apply(Cell cell) {
- return new ColumnWrapper(CellUtil.cloneFamily(cell), CellUtil.cloneQualifier(cell));
- }
-
- };
-
- return Multimaps.index(Iterables.filter(rawCells, shadowCellFilter), cellToColumnWrapper)
- .asMap().values()
- .asList();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/hbase-client/src/main/java/com/yahoo/omid/transaction/TTableCellGetterAdapter.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/com/yahoo/omid/transaction/TTableCellGetterAdapter.java b/hbase-client/src/main/java/com/yahoo/omid/transaction/TTableCellGetterAdapter.java
deleted file mode 100644
index 77548a5..0000000
--- a/hbase-client/src/main/java/com/yahoo/omid/transaction/TTableCellGetterAdapter.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package com.yahoo.omid.transaction;
-
-import com.yahoo.omid.transaction.CellUtils.CellGetter;
-import org.apache.hadoop.hbase.client.Get;
-import org.apache.hadoop.hbase.client.Result;
-
-import java.io.IOException;
-
-public class TTableCellGetterAdapter implements CellGetter {
-
- private final TTable txTable;
-
- public TTableCellGetterAdapter(TTable txTable) {
- this.txTable = txTable;
- }
-
- public Result get(Get get) throws IOException {
- return txTable.getHTable().get(get);
- }
-
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/hbase-client/src/main/java/org/apache/omid/transaction/ColumnWrapper.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/omid/transaction/ColumnWrapper.java b/hbase-client/src/main/java/org/apache/omid/transaction/ColumnWrapper.java
new file mode 100644
index 0000000..012d52e
--- /dev/null
+++ b/hbase-client/src/main/java/org/apache/omid/transaction/ColumnWrapper.java
@@ -0,0 +1,64 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.omid.transaction;
+
+import java.util.Arrays;
+
+public class ColumnWrapper {
+ private byte[] family;
+ private byte[] qualifier;
+
+ public ColumnWrapper(byte[] family, byte[] qualifier) {
+ this.family = family;
+ this.qualifier = qualifier;
+ }
+
+ public byte[] getFamily() {
+ return family;
+ }
+
+ public byte[] getQualifier() {
+ return qualifier;
+ }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + Arrays.hashCode(family);
+ result = prime * result + Arrays.hashCode(qualifier);
+ return result;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj)
+ return true;
+ if (obj == null)
+ return false;
+ if (getClass() != obj.getClass())
+ return false;
+ ColumnWrapper other = (ColumnWrapper) obj;
+ if (!Arrays.equals(family, other.family))
+ return false;
+ if (!Arrays.equals(qualifier, other.qualifier))
+ return false;
+ return true;
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/hbase-client/src/main/java/org/apache/omid/transaction/HBaseAsyncPostCommitter.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/omid/transaction/HBaseAsyncPostCommitter.java b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseAsyncPostCommitter.java
new file mode 100644
index 0000000..1037165
--- /dev/null
+++ b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseAsyncPostCommitter.java
@@ -0,0 +1,67 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.omid.transaction;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import org.apache.omid.tso.client.CellId;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Callable;
+
+public class HBaseAsyncPostCommitter implements PostCommitActions {
+
+ private static final Logger LOG = LoggerFactory.getLogger(HBaseAsyncPostCommitter.class);
+
+ private PostCommitActions syncPostCommitter;
+
+ private ListeningExecutorService postCommitExecutor;
+
+ public HBaseAsyncPostCommitter(PostCommitActions postCommitter, ListeningExecutorService postCommitExecutor) {
+ this.syncPostCommitter = postCommitter;
+ this.postCommitExecutor = postCommitExecutor;
+ }
+
+ @Override
+ public ListenableFuture<Void> updateShadowCells(final AbstractTransaction<? extends CellId> transaction) {
+
+ return postCommitExecutor.submit(new Callable<Void>() {
+ @Override
+ public Void call() throws Exception {
+ syncPostCommitter.updateShadowCells(transaction);
+ return null;
+ }
+
+ });
+
+ }
+
+ @Override
+ public ListenableFuture<Void> removeCommitTableEntry(final AbstractTransaction<? extends CellId> transaction) {
+
+ return postCommitExecutor.submit(new Callable<Void>() {
+ @Override
+ public Void call() throws Exception {
+ syncPostCommitter.removeCommitTableEntry(transaction);
+ return null;
+ }
+ });
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/hbase-client/src/main/java/org/apache/omid/transaction/HBaseCellId.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/omid/transaction/HBaseCellId.java b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseCellId.java
new file mode 100644
index 0000000..1f4add1
--- /dev/null
+++ b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseCellId.java
@@ -0,0 +1,80 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.omid.transaction;
+
+import com.google.common.hash.Hashing;
+import org.apache.omid.tso.client.CellId;
+import org.apache.hadoop.hbase.client.HTableInterface;
+
+import static com.google.common.base.Charsets.UTF_8;
+
+public class HBaseCellId implements CellId {
+
+ private final HTableInterface table;
+ private final byte[] row;
+ private final byte[] family;
+ private final byte[] qualifier;
+ private long timestamp;
+
+ public HBaseCellId(HTableInterface table, byte[] row, byte[] family, byte[] qualifier, long timestamp) {
+ this.timestamp = timestamp;
+ this.table = table;
+ this.row = row;
+ this.family = family;
+ this.qualifier = qualifier;
+ }
+
+ public HTableInterface getTable() {
+ return table;
+ }
+
+ public byte[] getRow() {
+ return row;
+ }
+
+ public byte[] getFamily() {
+ return family;
+ }
+
+ public byte[] getQualifier() {
+ return qualifier;
+ }
+
+ public long getTimestamp() {
+ return timestamp;
+ }
+
+ public String toString() {
+ return new String(table.getTableName(), UTF_8)
+ + ":" + new String(row, UTF_8)
+ + ":" + new String(family, UTF_8)
+ + ":" + new String(qualifier, UTF_8)
+ + ":" + timestamp;
+ }
+
+ @Override
+ public long getCellId() {
+ return Hashing.murmur3_128().newHasher()
+ .putBytes(table.getTableName())
+ .putBytes(row)
+ .putBytes(family)
+ .putBytes(qualifier)
+ .hash().asLong();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/hbase-client/src/main/java/org/apache/omid/transaction/HBaseOmidClientConfiguration.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/omid/transaction/HBaseOmidClientConfiguration.java b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseOmidClientConfiguration.java
new file mode 100644
index 0000000..822f8f5
--- /dev/null
+++ b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseOmidClientConfiguration.java
@@ -0,0 +1,170 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.omid.transaction;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.inject.Inject;
+import com.google.inject.name.Named;
+import org.apache.omid.YAMLUtils;
+import org.apache.omid.metrics.MetricsRegistry;
+import org.apache.omid.tools.hbase.SecureHBaseConfig;
+import org.apache.omid.tso.client.OmidClientConfiguration.PostCommitMode;
+import org.apache.omid.tso.client.OmidClientConfiguration;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+
+/**
+ * Configuration for HBase's Omid client side
+ */
+public class HBaseOmidClientConfiguration extends SecureHBaseConfig {
+
+ private static final String DEFAULT_CONFIG_FILE_NAME = "default-hbase-omid-client-config.yml";
+ private static final String CONFIG_FILE_NAME = "hbase-omid-client-config.yml";
+ private Configuration hbaseConfiguration = HBaseConfiguration.create();
+ private String commitTableName;
+ @Inject
+ private OmidClientConfiguration omidClientConfiguration;
+ private MetricsRegistry metrics;
+
+ // ----------------------------------------------------------------------------------------------------------------
+ // Instantiation
+ // ----------------------------------------------------------------------------------------------------------------
+ public HBaseOmidClientConfiguration() {
+ this(CONFIG_FILE_NAME);
+ }
+
+ @VisibleForTesting
+ HBaseOmidClientConfiguration(String configFileName) {
+ new YAMLUtils().loadSettings(configFileName, DEFAULT_CONFIG_FILE_NAME, this);
+ }
+
+ // ----------------------------------------------------------------------------------------------------------------
+ // Getters and setters for config params
+ // ----------------------------------------------------------------------------------------------------------------
+
+ public Configuration getHBaseConfiguration() {
+ return hbaseConfiguration;
+ }
+
+ public void setHBaseConfiguration(Configuration hbaseConfiguration) {
+ this.hbaseConfiguration = hbaseConfiguration;
+ }
+
+ public PostCommitMode getPostCommitMode() {
+ return omidClientConfiguration.getPostCommitMode();
+ }
+
+ public void setPostCommitMode(PostCommitMode postCommitMode) {
+ omidClientConfiguration.setPostCommitMode(postCommitMode);
+ }
+
+ public String getCommitTableName() {
+ return commitTableName;
+ }
+
+ @Inject(optional = true)
+ @Named("omid.client.hbase.commitTableName")
+ public void setCommitTableName(String commitTableName) {
+ this.commitTableName = commitTableName;
+ }
+
+ public OmidClientConfiguration getOmidClientConfiguration() {
+ return omidClientConfiguration;
+ }
+
+ public void setOmidClientConfiguration(OmidClientConfiguration omidClientConfiguration) {
+ this.omidClientConfiguration = omidClientConfiguration;
+ }
+
+ public MetricsRegistry getMetrics() {
+ return metrics;
+ }
+
+ @Inject(optional = true)
+ @Named("omid.client.hbase.metrics")
+ public void setMetrics(MetricsRegistry metrics) {
+ this.metrics = metrics;
+ }
+
+ // Delegation to make end-user life better
+
+ public OmidClientConfiguration.ConnType getConnectionType() {
+ return omidClientConfiguration.getConnectionType();
+ }
+
+ public void setReconnectionDelayInSecs(int reconnectionDelayInSecs) {
+ omidClientConfiguration.setReconnectionDelayInSecs(reconnectionDelayInSecs);
+ }
+
+ public void setExecutorThreads(int executorThreads) {
+ omidClientConfiguration.setExecutorThreads(executorThreads);
+ }
+
+ public int getRequestTimeoutInMs() {
+ return omidClientConfiguration.getRequestTimeoutInMs();
+ }
+
+ public void setConnectionString(String connectionString) {
+ omidClientConfiguration.setConnectionString(connectionString);
+ }
+
+ public void setRequestTimeoutInMs(int requestTimeoutInMs) {
+ omidClientConfiguration.setRequestTimeoutInMs(requestTimeoutInMs);
+ }
+
+ public void setZkConnectionTimeoutInSecs(int zkConnectionTimeoutInSecs) {
+ omidClientConfiguration.setZkConnectionTimeoutInSecs(zkConnectionTimeoutInSecs);
+ }
+
+ public void setConnectionType(OmidClientConfiguration.ConnType connectionType) {
+ omidClientConfiguration.setConnectionType(connectionType);
+ }
+
+ public void setRequestMaxRetries(int requestMaxRetries) {
+ omidClientConfiguration.setRequestMaxRetries(requestMaxRetries);
+ }
+
+ public int getZkConnectionTimeoutInSecs() {
+ return omidClientConfiguration.getZkConnectionTimeoutInSecs();
+ }
+
+ public void setRetryDelayInMs(int retryDelayInMs) {
+ omidClientConfiguration.setRetryDelayInMs(retryDelayInMs);
+ }
+
+ public int getExecutorThreads() {
+ return omidClientConfiguration.getExecutorThreads();
+ }
+
+ public int getRetryDelayInMs() {
+ return omidClientConfiguration.getRetryDelayInMs();
+ }
+
+ public String getConnectionString() {
+ return omidClientConfiguration.getConnectionString();
+ }
+
+ public int getRequestMaxRetries() {
+ return omidClientConfiguration.getRequestMaxRetries();
+ }
+
+ public int getReconnectionDelayInSecs() {
+ return omidClientConfiguration.getReconnectionDelayInSecs();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/hbase-client/src/main/java/org/apache/omid/transaction/HBaseSyncPostCommitter.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/omid/transaction/HBaseSyncPostCommitter.java b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseSyncPostCommitter.java
new file mode 100644
index 0000000..ba657fd
--- /dev/null
+++ b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseSyncPostCommitter.java
@@ -0,0 +1,125 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.omid.transaction;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
+import org.apache.omid.committable.CommitTable;
+import org.apache.omid.metrics.MetricsRegistry;
+import org.apache.omid.metrics.Timer;
+import org.apache.omid.tso.client.CellId;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.concurrent.ExecutionException;
+
+import static org.apache.omid.metrics.MetricsUtils.name;
+
+public class HBaseSyncPostCommitter implements PostCommitActions {
+
+ private static final Logger LOG = LoggerFactory.getLogger(HBaseSyncPostCommitter.class);
+
+ private final MetricsRegistry metrics;
+ private final CommitTable.Client commitTableClient;
+
+ private final Timer commitTableUpdateTimer;
+ private final Timer shadowCellsUpdateTimer;
+
+ public HBaseSyncPostCommitter(MetricsRegistry metrics, CommitTable.Client commitTableClient) {
+ this.metrics = metrics;
+ this.commitTableClient = commitTableClient;
+
+ this.commitTableUpdateTimer = metrics.timer(name("omid", "tm", "hbase", "commitTableUpdate", "latency"));
+ this.shadowCellsUpdateTimer = metrics.timer(name("omid", "tm", "hbase", "shadowCellsUpdate", "latency"));
+ }
+
+ @Override
+ public ListenableFuture<Void> updateShadowCells(AbstractTransaction<? extends CellId> transaction) {
+
+ SettableFuture<Void> updateSCFuture = SettableFuture.create();
+
+ HBaseTransaction tx = HBaseTransactionManager.enforceHBaseTransactionAsParam(transaction);
+
+ shadowCellsUpdateTimer.start();
+ try {
+
+ // Add shadow cells
+ for (HBaseCellId cell : tx.getWriteSet()) {
+ Put put = new Put(cell.getRow());
+ put.add(cell.getFamily(),
+ CellUtils.addShadowCellSuffix(cell.getQualifier(), 0, cell.getQualifier().length),
+ tx.getStartTimestamp(),
+ Bytes.toBytes(tx.getCommitTimestamp()));
+ try {
+ cell.getTable().put(put);
+ } catch (IOException e) {
+ LOG.warn("{}: Error inserting shadow cell {}", tx, cell, e);
+ updateSCFuture.setException(
+ new TransactionManagerException(tx + ": Error inserting shadow cell " + cell, e));
+ }
+ }
+
+ // Flush affected tables before returning to avoid loss of shadow cells updates when autoflush is disabled
+ try {
+ tx.flushTables();
+ updateSCFuture.set(null);
+ } catch (IOException e) {
+ LOG.warn("{}: Error while flushing writes", tx, e);
+ updateSCFuture.setException(new TransactionManagerException(tx + ": Error while flushing writes", e));
+ }
+
+ } finally {
+ shadowCellsUpdateTimer.stop();
+ }
+
+ return updateSCFuture;
+
+ }
+
+ @Override
+ public ListenableFuture<Void> removeCommitTableEntry(AbstractTransaction<? extends CellId> transaction) {
+
+ SettableFuture<Void> updateSCFuture = SettableFuture.create();
+
+ HBaseTransaction tx = HBaseTransactionManager.enforceHBaseTransactionAsParam(transaction);
+
+ commitTableUpdateTimer.start();
+
+ try {
+ commitTableClient.completeTransaction(tx.getStartTimestamp()).get();
+ updateSCFuture.set(null);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOG.warn("{}: interrupted during commit table entry delete", tx, e);
+ updateSCFuture.setException(
+ new TransactionManagerException(tx + ": interrupted during commit table entry delete"));
+ } catch (ExecutionException e) {
+ LOG.warn("{}: can't remove commit table entry", tx, e);
+ updateSCFuture.setException(new TransactionManagerException(tx + ": can't remove commit table entry"));
+ } finally {
+ commitTableUpdateTimer.stop();
+ }
+
+ return updateSCFuture;
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransaction.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransaction.java b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransaction.java
new file mode 100644
index 0000000..6e00de0
--- /dev/null
+++ b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransaction.java
@@ -0,0 +1,79 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.omid.transaction;
+
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.HTableInterface;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
+
+public class HBaseTransaction extends AbstractTransaction<HBaseCellId> {
+ private static final Logger LOG = LoggerFactory.getLogger(HBaseTransaction.class);
+
+ HBaseTransaction(long transactionId, long epoch, Set<HBaseCellId> writeSet, AbstractTransactionManager tm) {
+ super(transactionId, epoch, writeSet, tm);
+ }
+
+ @Override
+ public void cleanup() {
+ Set<HBaseCellId> writeSet = getWriteSet();
+ for (final HBaseCellId cell : writeSet) {
+ Delete delete = new Delete(cell.getRow());
+ delete.deleteColumn(cell.getFamily(), cell.getQualifier(), getStartTimestamp());
+ try {
+ cell.getTable().delete(delete);
+ } catch (IOException e) {
+ LOG.warn("Failed cleanup cell {} for Tx {}. This issue has been ignored", new Object[]{cell, getTransactionId(), e});
+ }
+ }
+ try {
+ flushTables();
+ } catch (IOException e) {
+ LOG.warn("Failed flushing tables for Tx {}", getTransactionId(), e);
+ }
+ }
+
+ /**
+ * Flushes pending operations for tables touched by transaction
+ */
+ public void flushTables() throws IOException {
+
+ for (HTableInterface writtenTable : getWrittenTables()) {
+ writtenTable.flushCommits();
+ }
+
+ }
+
+ // ****************************************************************************************************************
+ // Helper methods
+ // ****************************************************************************************************************
+
+ private Set<HTableInterface> getWrittenTables() {
+ HashSet<HBaseCellId> writeSet = (HashSet<HBaseCellId>) getWriteSet();
+ Set<HTableInterface> tables = new HashSet<HTableInterface>();
+ for (HBaseCellId cell : writeSet) {
+ tables.add(cell.getTable());
+ }
+ return tables;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransactionClient.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransactionClient.java b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransactionClient.java
new file mode 100644
index 0000000..aaf236b
--- /dev/null
+++ b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransactionClient.java
@@ -0,0 +1,24 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.omid.transaction;
+
+public interface HBaseTransactionClient {
+ boolean isCommitted(HBaseCellId hBaseCellId) throws TransactionException;
+
+ long getLowWatermark() throws TransactionException;
+}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransactionManager.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransactionManager.java b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransactionManager.java
new file mode 100644
index 0000000..fd45dc2
--- /dev/null
+++ b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransactionManager.java
@@ -0,0 +1,292 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.omid.transaction;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Optional;
+import com.google.common.collect.Maps;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.omid.committable.CommitTable;
+import org.apache.omid.committable.CommitTable.CommitTimestamp;
+import org.apache.omid.committable.hbase.HBaseCommitTable;
+import org.apache.omid.committable.hbase.HBaseCommitTableConfig;
+import org.apache.omid.tools.hbase.HBaseLogin;
+import org.apache.omid.tso.client.CellId;
+import org.apache.omid.tso.client.TSOClient;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+
+public class HBaseTransactionManager extends AbstractTransactionManager implements HBaseTransactionClient {
+
+ private static final Logger LOG = LoggerFactory.getLogger(HBaseTransactionManager.class);
+
+ private static class HBaseTransactionFactory implements TransactionFactory<HBaseCellId> {
+
+ @Override
+ public HBaseTransaction createTransaction(long transactionId, long epoch, AbstractTransactionManager tm) {
+
+ return new HBaseTransaction(transactionId, epoch, new HashSet<HBaseCellId>(), tm);
+
+ }
+
+ }
+
+ // ----------------------------------------------------------------------------------------------------------------
+ // Construction
+ // ----------------------------------------------------------------------------------------------------------------
+
+ public static TransactionManager newInstance() throws IOException, InterruptedException {
+ return newInstance(new HBaseOmidClientConfiguration());
+ }
+
+ public static TransactionManager newInstance(HBaseOmidClientConfiguration configuration)
+ throws IOException, InterruptedException {
+ //Logging in to Secure HBase if required
+ HBaseLogin.loginIfNeeded(configuration);
+ return builder(configuration).build();
+ }
+
+ @VisibleForTesting
+ static class Builder {
+
+ // Required parameters
+ private final HBaseOmidClientConfiguration hbaseOmidClientConf;
+
+ // Optional parameters - initialized to default values
+ private Optional<TSOClient> tsoClient = Optional.absent();
+ private Optional<CommitTable.Client> commitTableClient = Optional.absent();
+ private Optional<PostCommitActions> postCommitter = Optional.absent();
+
+ private Builder(HBaseOmidClientConfiguration hbaseOmidClientConf) {
+ this.hbaseOmidClientConf = hbaseOmidClientConf;
+ }
+
+ Builder tsoClient(TSOClient tsoClient) {
+ this.tsoClient = Optional.of(tsoClient);
+ return this;
+ }
+
+ Builder commitTableClient(CommitTable.Client client) {
+ this.commitTableClient = Optional.of(client);
+ return this;
+ }
+
+ Builder postCommitter(PostCommitActions postCommitter) {
+ this.postCommitter = Optional.of(postCommitter);
+ return this;
+ }
+
+ HBaseTransactionManager build() throws IOException, InterruptedException {
+
+ CommitTable.Client commitTableClient = this.commitTableClient.or(buildCommitTableClient()).get();
+ PostCommitActions postCommitter = this.postCommitter.or(buildPostCommitter(commitTableClient)).get();
+ TSOClient tsoClient = this.tsoClient.or(buildTSOClient()).get();
+
+ return new HBaseTransactionManager(hbaseOmidClientConf,
+ postCommitter,
+ tsoClient,
+ commitTableClient,
+ new HBaseTransactionFactory());
+ }
+
+ private Optional<TSOClient> buildTSOClient() throws IOException, InterruptedException {
+ return Optional.of(TSOClient.newInstance(hbaseOmidClientConf.getOmidClientConfiguration()));
+ }
+
+
+ private Optional<CommitTable.Client> buildCommitTableClient() throws IOException {
+ HBaseCommitTableConfig commitTableConf = new HBaseCommitTableConfig();
+ commitTableConf.setTableName(hbaseOmidClientConf.getCommitTableName());
+ CommitTable commitTable = new HBaseCommitTable(hbaseOmidClientConf.getHBaseConfiguration(), commitTableConf);
+ return Optional.of(commitTable.getClient());
+ }
+
+ private Optional<PostCommitActions> buildPostCommitter(CommitTable.Client commitTableClient ) {
+
+ PostCommitActions postCommitter;
+ PostCommitActions syncPostCommitter = new HBaseSyncPostCommitter(hbaseOmidClientConf.getMetrics(),
+ commitTableClient);
+ switch(hbaseOmidClientConf.getPostCommitMode()) {
+ case ASYNC:
+ ListeningExecutorService postCommitExecutor =
+ MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor(
+ new ThreadFactoryBuilder().setNameFormat("postCommit-%d").build()));
+ postCommitter = new HBaseAsyncPostCommitter(syncPostCommitter, postCommitExecutor);
+ break;
+ case SYNC:
+ default:
+ postCommitter = syncPostCommitter;
+ break;
+ }
+
+ return Optional.of(postCommitter);
+ }
+
+ }
+
+ @VisibleForTesting
+ static Builder builder(HBaseOmidClientConfiguration hbaseOmidClientConf) {
+ return new Builder(hbaseOmidClientConf);
+ }
+
+ private HBaseTransactionManager(HBaseOmidClientConfiguration hBaseOmidClientConfiguration,
+ PostCommitActions postCommitter,
+ TSOClient tsoClient,
+ CommitTable.Client commitTableClient,
+ HBaseTransactionFactory hBaseTransactionFactory) {
+
+ super(hBaseOmidClientConfiguration.getMetrics(),
+ postCommitter,
+ tsoClient,
+ commitTableClient,
+ hBaseTransactionFactory);
+
+ }
+
+ // ----------------------------------------------------------------------------------------------------------------
+ // AbstractTransactionManager overwritten methods
+ // ----------------------------------------------------------------------------------------------------------------
+
+ @Override
+ public void preCommit(AbstractTransaction<? extends CellId> transaction) throws TransactionManagerException {
+ try {
+ // Flush all pending writes
+ HBaseTransaction hBaseTx = enforceHBaseTransactionAsParam(transaction);
+ hBaseTx.flushTables();
+ } catch (IOException e) {
+ throw new TransactionManagerException("Exception while flushing writes", e);
+ }
+ }
+
+ @Override
+ public void preRollback(AbstractTransaction<? extends CellId> transaction) throws TransactionManagerException {
+ try {
+ // Flush all pending writes
+ HBaseTransaction hBaseTx = enforceHBaseTransactionAsParam(transaction);
+ hBaseTx.flushTables();
+ } catch (IOException e) {
+ throw new TransactionManagerException("Exception while flushing writes", e);
+ }
+ }
+
+ // ----------------------------------------------------------------------------------------------------------------
+ // HBaseTransactionClient method implementations
+ // ----------------------------------------------------------------------------------------------------------------
+
+ @Override
+ public boolean isCommitted(HBaseCellId hBaseCellId) throws TransactionException {
+ try {
+ CommitTimestamp tentativeCommitTimestamp =
+ locateCellCommitTimestamp(hBaseCellId.getTimestamp(), tsoClient.getEpoch(),
+ new CommitTimestampLocatorImpl(hBaseCellId, Maps.<Long, Long>newHashMap()));
+
+ // If transaction that added the cell was invalidated
+ if (!tentativeCommitTimestamp.isValid()) {
+ return false;
+ }
+
+ switch (tentativeCommitTimestamp.getLocation()) {
+ case COMMIT_TABLE:
+ case SHADOW_CELL:
+ return true;
+ case NOT_PRESENT:
+ return false;
+ case CACHE: // cache was empty
+ default:
+ return false;
+ }
+ } catch (IOException e) {
+ throw new TransactionException("Failure while checking if a transaction was committed", e);
+ }
+ }
+
+ @Override
+ public long getLowWatermark() throws TransactionException {
+ try {
+ return commitTableClient.readLowWatermark().get();
+ } catch (ExecutionException ee) {
+ throw new TransactionException("Error reading low watermark", ee.getCause());
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ throw new TransactionException("Interrupted reading low watermark", ie);
+ }
+ }
+
+ // ----------------------------------------------------------------------------------------------------------------
+ // Helper methods
+ // ----------------------------------------------------------------------------------------------------------------
+
+ static HBaseTransaction enforceHBaseTransactionAsParam(AbstractTransaction<? extends CellId> tx) {
+
+ if (tx instanceof HBaseTransaction) {
+ return (HBaseTransaction) tx;
+ } else {
+ throw new IllegalArgumentException(
+ "The transaction object passed is not an instance of HBaseTransaction");
+ }
+
+ }
+
+ static class CommitTimestampLocatorImpl implements CommitTimestampLocator {
+
+ private HBaseCellId hBaseCellId;
+ private final Map<Long, Long> commitCache;
+
+ CommitTimestampLocatorImpl(HBaseCellId hBaseCellId, Map<Long, Long> commitCache) {
+ this.hBaseCellId = hBaseCellId;
+ this.commitCache = commitCache;
+ }
+
+ @Override
+ public Optional<Long> readCommitTimestampFromCache(long startTimestamp) {
+ if (commitCache.containsKey(startTimestamp)) {
+ return Optional.of(commitCache.get(startTimestamp));
+ }
+ return Optional.absent();
+ }
+
+ @Override
+ public Optional<Long> readCommitTimestampFromShadowCell(long startTimestamp) throws IOException {
+
+ Get get = new Get(hBaseCellId.getRow());
+ byte[] family = hBaseCellId.getFamily();
+ byte[] shadowCellQualifier = CellUtils.addShadowCellSuffix(hBaseCellId.getQualifier());
+ get.addColumn(family, shadowCellQualifier);
+ get.setMaxVersions(1);
+ get.setTimeStamp(startTimestamp);
+ Result result = hBaseCellId.getTable().get(get);
+ if (result.containsColumn(family, shadowCellQualifier)) {
+ return Optional.of(Bytes.toLong(result.getValue(family, shadowCellQualifier)));
+ }
+ return Optional.absent();
+ }
+
+ }
+
+}