You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tephra.apache.org by po...@apache.org on 2016/05/11 20:15:49 UTC
[49/56] [abbrv] [partial] incubator-tephra git commit: Rename package
to org.apache.tephra
http://git-wip-us.apache.org/repos/asf/incubator-tephra/blob/9c693743/tephra-core/src/main/java/co/cask/tephra/TransactionManager.java
----------------------------------------------------------------------
diff --git a/tephra-core/src/main/java/co/cask/tephra/TransactionManager.java b/tephra-core/src/main/java/co/cask/tephra/TransactionManager.java
deleted file mode 100644
index 3f51956..0000000
--- a/tephra-core/src/main/java/co/cask/tephra/TransactionManager.java
+++ /dev/null
@@ -1,1398 +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 co.cask.tephra;
-
-import co.cask.tephra.metrics.DefaultMetricsCollector;
-import co.cask.tephra.metrics.MetricsCollector;
-import co.cask.tephra.persist.NoOpTransactionStateStorage;
-import co.cask.tephra.persist.TransactionEdit;
-import co.cask.tephra.persist.TransactionLog;
-import co.cask.tephra.persist.TransactionLogReader;
-import co.cask.tephra.persist.TransactionSnapshot;
-import co.cask.tephra.persist.TransactionStateStorage;
-import co.cask.tephra.snapshot.SnapshotCodecProvider;
-import co.cask.tephra.util.TxUtils;
-import com.google.common.base.Objects;
-import com.google.common.base.Preconditions;
-import com.google.common.base.Stopwatch;
-import com.google.common.base.Throwables;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
-import com.google.common.util.concurrent.AbstractService;
-import com.google.inject.Inject;
-import it.unimi.dsi.fastutil.longs.LongArrayList;
-import org.apache.hadoop.conf.Configuration;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.io.OutputStream;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.NavigableMap;
-import java.util.Set;
-import java.util.concurrent.ConcurrentSkipListMap;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-import javax.annotation.Nonnull;
-import javax.annotation.Nullable;
-
-/**
- * This is the central place to manage all active transactions in the system.
- *
- * A transaction consists of
- * <ul>
- * <li>A write pointer: This is the version used for all writes of that transaction.</li>
- * <li>A read pointer: All reads under the transaction use this as an upper bound for the version.</li>
- * <li>A set of excluded versions: These are the write versions of other transactions that must be excluded from
- * reads, because those transactions are still in progress, or they failed but couldn't be properly rolled back.</li>
- * </ul>
- * To use the transaction system, a client must follow this sequence of steps:
- * <ol>
- * <li>Request a new transaction.</li>
- * <li>Use the transaction to read and write datasets. Datasets are encouraged to cache the writes of the
- * transaction in memory, to reduce the cost of rollback in case the transaction fails. </li>
- * <li>Check whether the transaction has conflicts. For this, the set of change keys are submitted via canCommit(),
- * and the transaction manager verifies that none of these keys are in conflict with other transactions that
- * committed since the start of this transaction.</li>
- * <li>If the transaction has conflicts:
- * <ol>
- * <li>Roll back the changes in every dataset that was changed. This can happen in-memory if the
- * changes were cached.</li>
- * <li>Abort the transaction to remove it from the active transactions.</li>
- * </ol>
- * <li>If the transaction has no conflicts:</li>
- * <ol>
- * <li>Persist all datasets changes to storage.</li>
- * <li>Commit the transaction. This will repeat the conflict detection, because more overlapping transactions
- * may have committed since the first conflict check.</li>
- * <li>If the transaction has conflicts:</li>
- * <ol>
- * <li>Roll back the changes in every dataset that was changed. This is more expensive because
- * changes must be undone in persistent storage.</li>
- * <li>Abort the transaction to remove it from the active transactions.</li>
- * </ol>
- * </ol>
- * </ol>
- * Transactions may be short or long-running. A short transaction is started with a timeout, and if it is not
- * committed before that timeout, it is invalidated and excluded from future reads. A long-running transaction has
- * no timeout and will remain active until it is committed or aborted. Long transactions are typically used in
- * map/reduce jobs and can produce enormous amounts of changes. Therefore, long transactions do not participate in
- * conflict detection (they would almost always have conflicts). We also assume that the changes of long transactions
- * are not tracked, and therefore cannot be rolled back. Hence, when a long transaction is aborted, it remains in the
- * list of excluded transactions to make its writes invisible.
- */
-public class TransactionManager extends AbstractService {
- // todo: optimize heavily
-
- private static final Logger LOG = LoggerFactory.getLogger(TransactionManager.class);
-
- // poll every 1 second to check whether a snapshot is needed
- private static final long SNAPSHOT_POLL_INTERVAL = 1000L;
-
- //poll every 10 second to emit metrics
- private static final long METRICS_POLL_INTERVAL = 10000L;
-
- private static final long[] NO_INVALID_TX = { };
-
- // Transactions that are in progress, with their info.
- private final NavigableMap<Long, InProgressTx> inProgress = new ConcurrentSkipListMap<Long, InProgressTx>();
-
- // the list of transactions that are invalid (not properly committed/aborted, or timed out)
- // TODO: explain usage of two arrays
- private final LongArrayList invalid = new LongArrayList();
- private long[] invalidArray = NO_INVALID_TX;
-
- // todo: use moving array instead (use Long2ObjectMap<byte[]> in fastutil)
- // todo: should this be consolidated with inProgress?
- // commit time next writePointer -> changes made by this tx
- private final NavigableMap<Long, Set<ChangeId>> committedChangeSets =
- new ConcurrentSkipListMap<Long, Set<ChangeId>>();
- // not committed yet
- private final Map<Long, Set<ChangeId>> committingChangeSets = Maps.newConcurrentMap();
-
- private long readPointer;
- private long lastWritePointer;
- private MetricsCollector txMetricsCollector;
-
- private final TransactionStateStorage persistor;
-
- private final int cleanupInterval;
- private final int defaultTimeout;
- private final int defaultLongTimeout;
- private DaemonThreadExecutor cleanupThread = null;
-
- private volatile TransactionLog currentLog;
-
- // timestamp of the last completed snapshot
- private long lastSnapshotTime;
- // frequency in millis to perform snapshots
- private final long snapshotFrequencyInSeconds;
- // number of most recent snapshots to retain
- private final int snapshotRetainCount;
- private DaemonThreadExecutor snapshotThread;
- private DaemonThreadExecutor metricsThread;
-
- // lock guarding change of the current transaction log
- private final ReentrantReadWriteLock logLock = new ReentrantReadWriteLock();
- private final Lock logReadLock = logLock.readLock();
- private final Lock logWriteLock = logLock.writeLock();
-
- // fudge factor (in milliseconds) used when interpreting transactions as LONG based on expiration
- // TODO: REMOVE WITH txnBackwardsCompatCheck()
- private final long longTimeoutTolerance;
-
- public TransactionManager(Configuration config) {
- this(config, new NoOpTransactionStateStorage(new SnapshotCodecProvider(config)), new DefaultMetricsCollector());
- }
-
- @Inject
- public TransactionManager(Configuration conf, @Nonnull TransactionStateStorage persistor,
- MetricsCollector txMetricsCollector) {
- this.persistor = persistor;
- cleanupInterval = conf.getInt(TxConstants.Manager.CFG_TX_CLEANUP_INTERVAL,
- TxConstants.Manager.DEFAULT_TX_CLEANUP_INTERVAL);
- defaultTimeout = conf.getInt(TxConstants.Manager.CFG_TX_TIMEOUT,
- TxConstants.Manager.DEFAULT_TX_TIMEOUT);
- defaultLongTimeout = conf.getInt(TxConstants.Manager.CFG_TX_LONG_TIMEOUT,
- TxConstants.Manager.DEFAULT_TX_LONG_TIMEOUT);
- snapshotFrequencyInSeconds = conf.getLong(TxConstants.Manager.CFG_TX_SNAPSHOT_INTERVAL,
- TxConstants.Manager.DEFAULT_TX_SNAPSHOT_INTERVAL);
- // must always keep at least 1 snapshot
- snapshotRetainCount = Math.max(conf.getInt(TxConstants.Manager.CFG_TX_SNAPSHOT_RETAIN,
- TxConstants.Manager.DEFAULT_TX_SNAPSHOT_RETAIN), 1);
-
- // intentionally not using a constant, as this config should not be exposed
- // TODO: REMOVE WITH txnBackwardsCompatCheck()
- longTimeoutTolerance = conf.getLong("data.tx.long.timeout.tolerance", 10000);
-
- //
- this.txMetricsCollector = txMetricsCollector;
- this.txMetricsCollector.configure(conf);
- clear();
- }
-
- private void clear() {
- invalid.clear();
- invalidArray = NO_INVALID_TX;
- inProgress.clear();
- committedChangeSets.clear();
- committingChangeSets.clear();
- lastWritePointer = 0;
- readPointer = 0;
- lastSnapshotTime = 0;
- }
-
- private boolean isStopping() {
- return State.STOPPING.equals(state());
- }
-
- @Override
- public synchronized void doStart() {
- LOG.info("Starting transaction manager.");
- txMetricsCollector.start();
- // start up the persistor
- persistor.startAndWait();
- try {
- persistor.setupStorage();
- } catch (IOException e) {
- Throwables.propagate(e);
- }
- // establish defaults in case there is no persistence
- clear();
- // attempt to recover state from last run
- recoverState();
- // start the periodic cleanup thread
- startCleanupThread();
- startSnapshotThread();
- startMetricsThread();
- // initialize the WAL if we did not force a snapshot in recoverState()
- initLog();
- // initialize next write pointer if needed
- if (lastWritePointer == 0) {
- lastWritePointer = getNextWritePointer();
- readPointer = lastWritePointer;
- }
-
- notifyStarted();
- }
-
- private void initLog() {
- if (currentLog == null) {
- try {
- currentLog = persistor.createLog(System.currentTimeMillis());
- } catch (IOException ioe) {
- throw Throwables.propagate(ioe);
- }
- }
- }
-
- private void startCleanupThread() {
- if (cleanupInterval <= 0 || defaultTimeout <= 0) {
- return;
- }
- LOG.info("Starting periodic timed-out transaction cleanup every " + cleanupInterval +
- " seconds with default timeout of " + defaultTimeout + " seconds.");
- this.cleanupThread = new DaemonThreadExecutor("tx-clean-timeout") {
- @Override
- public void doRun() {
- cleanupTimedOutTransactions();
- }
-
- @Override
- public long getSleepMillis() {
- return cleanupInterval * 1000;
- }
- };
- cleanupThread.start();
- }
-
- private void startSnapshotThread() {
- if (snapshotFrequencyInSeconds > 0) {
- LOG.info("Starting periodic snapshot thread, frequency = " + snapshotFrequencyInSeconds +
- " seconds, location = " + persistor.getLocation());
- this.snapshotThread = new DaemonThreadExecutor("tx-snapshot") {
- @Override
- public void doRun() {
- long currentTime = System.currentTimeMillis();
- if (lastSnapshotTime < (currentTime - snapshotFrequencyInSeconds * 1000)) {
- try {
- doSnapshot(false);
- } catch (IOException ioe) {
- LOG.error("Periodic snapshot failed!", ioe);
- }
- }
- }
-
- @Override
- protected void onShutdown() {
- // perform a final snapshot
- try {
- LOG.info("Writing final snapshot prior to shutdown");
- doSnapshot(true);
- } catch (IOException ioe) {
- LOG.error("Failed performing final snapshot on shutdown", ioe);
- }
- }
-
- @Override
- public long getSleepMillis() {
- return SNAPSHOT_POLL_INTERVAL;
- }
- };
- snapshotThread.start();
- }
- }
-
- // Emits Transaction Data structures size as metrics
- private void startMetricsThread() {
- LOG.info("Starting periodic Metrics Emitter thread, frequency = " + METRICS_POLL_INTERVAL);
- this.metricsThread = new DaemonThreadExecutor("tx-metrics") {
- @Override
- public void doRun() {
- txMetricsCollector.gauge("committing.size", committingChangeSets.size());
- txMetricsCollector.gauge("committed.size", committedChangeSets.size());
- txMetricsCollector.gauge("inprogress.size", inProgress.size());
- txMetricsCollector.gauge("invalid.size", invalidArray.length);
- }
-
- @Override
- protected void onShutdown() {
- // perform a final metrics emit
- txMetricsCollector.gauge("committing.size", committingChangeSets.size());
- txMetricsCollector.gauge("committed.size", committedChangeSets.size());
- txMetricsCollector.gauge("inprogress.size", inProgress.size());
- txMetricsCollector.gauge("invalid.size", invalidArray.length);
- }
-
- @Override
- public long getSleepMillis() {
- return METRICS_POLL_INTERVAL;
- }
- };
- metricsThread.start();
- }
-
- private void cleanupTimedOutTransactions() {
- List<TransactionEdit> invalidEdits = null;
- this.logReadLock.lock();
- try {
- synchronized (this) {
- if (!isRunning()) {
- return;
- }
-
- long currentTime = System.currentTimeMillis();
- List<Long> timedOut = Lists.newArrayList();
- for (Map.Entry<Long, InProgressTx> tx : inProgress.entrySet()) {
- long expiration = tx.getValue().getExpiration();
- if (expiration >= 0L && currentTime > expiration) {
- // timed out, remember tx id (can't remove while iterating over entries)
- timedOut.add(tx.getKey());
- LOG.info("Tx invalid list: added tx {} because of timeout", tx.getKey());
- } else if (expiration < 0) {
- LOG.warn("Transaction {} has negative expiration time {}. Likely cause is the transaction was not " +
- "migrated correctly, this transaction will be expired immediately",
- tx.getKey(), expiration);
- timedOut.add(tx.getKey());
- }
- }
- if (!timedOut.isEmpty()) {
- invalidEdits = Lists.newArrayListWithCapacity(timedOut.size());
- invalid.addAll(timedOut);
- for (long tx : timedOut) {
- committingChangeSets.remove(tx);
- inProgress.remove(tx);
- invalidEdits.add(TransactionEdit.createInvalid(tx));
- }
-
- // todo: find a more efficient way to keep this sorted. Could it just be an array?
- Collections.sort(invalid);
- invalidArray = invalid.toLongArray();
- LOG.info("Invalidated {} transactions due to timeout.", timedOut.size());
- }
- }
- if (invalidEdits != null) {
- appendToLog(invalidEdits);
- }
- } finally {
- this.logReadLock.unlock();
- }
- }
-
- public synchronized TransactionSnapshot getSnapshot() throws IOException {
- TransactionSnapshot snapshot = null;
- if (!isRunning() && !isStopping()) {
- return null;
- }
-
- long now = System.currentTimeMillis();
- // avoid duplicate snapshots at same timestamp
- if (now == lastSnapshotTime || (currentLog != null && now == currentLog.getTimestamp())) {
- try {
- TimeUnit.MILLISECONDS.sleep(1);
- } catch (InterruptedException ie) { }
- }
- // copy in memory state
- snapshot = getCurrentState();
-
- LOG.debug("Starting snapshot of transaction state with timestamp {}", snapshot.getTimestamp());
- LOG.debug("Returning snapshot of state: " + snapshot);
- return snapshot;
- }
-
- /**
- * Take a snapshot of the transaction state and serialize it into the given output stream.
- * @return whether a snapshot was taken.
- */
- public boolean takeSnapshot(OutputStream out) throws IOException {
- TransactionSnapshot snapshot = getSnapshot();
- if (snapshot != null) {
- persistor.writeSnapshot(out, snapshot);
- return true;
- } else {
- return false;
- }
- }
-
- private void doSnapshot(boolean closing) throws IOException {
- long snapshotTime = 0L;
- TransactionSnapshot snapshot = null;
- TransactionLog oldLog = null;
- try {
- this.logWriteLock.lock();
- try {
- synchronized (this) {
- snapshot = getSnapshot();
- if (snapshot == null && !closing) {
- return;
- }
- if (snapshot != null) {
- snapshotTime = snapshot.getTimestamp();
- }
-
- // roll WAL
- oldLog = currentLog;
- if (!closing) {
- currentLog = persistor.createLog(snapshot.getTimestamp());
- }
- }
- // there may not be an existing log on startup
- if (oldLog != null) {
- oldLog.close();
- }
- } finally {
- this.logWriteLock.unlock();
- }
-
- // save snapshot
- if (snapshot != null) {
- persistor.writeSnapshot(snapshot);
- lastSnapshotTime = snapshotTime;
-
- // clean any obsoleted snapshots and WALs
- long oldestRetainedTimestamp = persistor.deleteOldSnapshots(snapshotRetainCount);
- persistor.deleteLogsOlderThan(oldestRetainedTimestamp);
- }
- } catch (IOException ioe) {
- abortService("Snapshot (timestamp " + snapshotTime + ") failed due to: " + ioe.getMessage(), ioe);
- }
- }
-
- public synchronized TransactionSnapshot getCurrentState() {
- return TransactionSnapshot.copyFrom(System.currentTimeMillis(), readPointer, lastWritePointer,
- invalid, inProgress, committingChangeSets, committedChangeSets);
- }
-
- public synchronized void recoverState() {
- try {
- TransactionSnapshot lastSnapshot = persistor.getLatestSnapshot();
- // if we failed before a snapshot could complete, we might not have one to restore
- if (lastSnapshot != null) {
- restoreSnapshot(lastSnapshot);
- }
- // replay any WALs since the last snapshot
- Collection<TransactionLog> logs = persistor.getLogsSince(lastSnapshotTime);
- if (logs != null) {
- replayLogs(logs);
- }
- } catch (IOException e) {
- LOG.error("Unable to read back transaction state:", e);
- throw Throwables.propagate(e);
- }
- }
-
- /**
- * Restore the initial in-memory transaction state from a snapshot.
- */
- private void restoreSnapshot(TransactionSnapshot snapshot) {
- LOG.info("Restoring transaction state from snapshot at " + snapshot.getTimestamp());
- Preconditions.checkState(lastSnapshotTime == 0, "lastSnapshotTime has been set!");
- Preconditions.checkState(readPointer == 0, "readPointer has been set!");
- Preconditions.checkState(lastWritePointer == 0, "lastWritePointer has been set!");
- Preconditions.checkState(invalid.isEmpty(), "invalid list should be empty!");
- Preconditions.checkState(inProgress.isEmpty(), "inProgress map should be empty!");
- Preconditions.checkState(committingChangeSets.isEmpty(), "committingChangeSets should be empty!");
- Preconditions.checkState(committedChangeSets.isEmpty(), "committedChangeSets should be empty!");
- LOG.info("Restoring snapshot of state: " + snapshot);
-
- lastSnapshotTime = snapshot.getTimestamp();
- readPointer = snapshot.getReadPointer();
- lastWritePointer = snapshot.getWritePointer();
- invalid.addAll(snapshot.getInvalid());
- inProgress.putAll(txnBackwardsCompatCheck(defaultLongTimeout, longTimeoutTolerance, snapshot.getInProgress()));
- committingChangeSets.putAll(snapshot.getCommittingChangeSets());
- committedChangeSets.putAll(snapshot.getCommittedChangeSets());
- }
-
- /**
- * Check if in-progress transactions need to be migrated to have expiration time and type, if so do the migration.
- * This is required for backwards compatibility, when long running transactions were represented
- * with expiration time -1. This can be removed when we stop supporting SnapshotCodec version 1.
- */
- public static Map<Long, InProgressTx> txnBackwardsCompatCheck(int defaultLongTimeout, long longTimeoutTolerance,
- Map<Long, InProgressTx> inProgress) {
- for (Map.Entry<Long, InProgressTx> entry : inProgress.entrySet()) {
- long writePointer = entry.getKey();
- long expiration = entry.getValue().getExpiration();
- // LONG transactions will either have a negative expiration or expiration set to the long timeout
- // use a fudge factor on the expiration check, since expiraton is set based on system time, not the write pointer
- if (entry.getValue().getType() == null &&
- (expiration < 0 ||
- (getTxExpirationFromWritePointer(writePointer, defaultLongTimeout) - expiration
- < longTimeoutTolerance))) {
- // handle null expiration
- long newExpiration = getTxExpirationFromWritePointer(writePointer, defaultLongTimeout);
- InProgressTx compatTx =
- new InProgressTx(entry.getValue().getVisibilityUpperBound(), newExpiration, TransactionType.LONG,
- entry.getValue().getCheckpointWritePointers());
- entry.setValue(compatTx);
- } else if (entry.getValue().getType() == null) {
- InProgressTx compatTx =
- new InProgressTx(entry.getValue().getVisibilityUpperBound(), entry.getValue().getExpiration(),
- TransactionType.SHORT, entry.getValue().getCheckpointWritePointers());
- entry.setValue(compatTx);
- }
- }
- return inProgress;
- }
-
- /**
- * Resets the state of the transaction manager.
- */
- public void resetState() {
- this.logWriteLock.lock();
- try {
- // Take a snapshot before resetting the state, for debugging purposes
- doSnapshot(false);
- // Clear the state
- clear();
- // Take another snapshot: if no snapshot is taken after clearing the state
- // and the manager is restarted, we will recover from the snapshot taken
- // before resetting the state, which would be really bad
- // This call will also init a new WAL
- doSnapshot(false);
- } catch (IOException e) {
- LOG.error("Snapshot failed when resetting state!", e);
- e.printStackTrace();
- } finally {
- this.logWriteLock.unlock();
- }
- }
-
- /**
- * Replay all logged edits from the given transaction logs.
- */
- private void replayLogs(Collection<TransactionLog> logs) {
- for (TransactionLog log : logs) {
- LOG.info("Replaying edits from transaction log " + log.getName());
- int editCnt = 0;
- try {
- TransactionLogReader reader = log.getReader();
- // reader may be null in the case of an empty file
- if (reader == null) {
- continue;
- }
- TransactionEdit edit = null;
- while ((edit = reader.next()) != null) {
- editCnt++;
- switch (edit.getState()) {
- case INPROGRESS:
- long expiration = edit.getExpiration();
- TransactionType type = edit.getType();
- // Check if transaction needs to be migrated to have expiration and type. Previous version of
- // long running transactions were represented with expiration time as -1.
- // This can be removed when we stop supporting TransactionEditCodecV2.
- if (expiration < 0) {
- expiration = getTxExpirationFromWritePointer(edit.getWritePointer(), defaultLongTimeout);
- type = TransactionType.LONG;
- } else if (type == null) {
- type = TransactionType.SHORT;
- }
- addInProgressAndAdvance(edit.getWritePointer(), edit.getVisibilityUpperBound(),
- expiration, type);
- break;
- case COMMITTING:
- addCommittingChangeSet(edit.getWritePointer(), edit.getChanges());
- break;
- case COMMITTED:
- // TODO: need to reconcile usage of transaction id v/s write pointer TEPHRA-140
- long transactionId = edit.getWritePointer();
- long[] checkpointPointers = edit.getCheckpointPointers();
- long writePointer = checkpointPointers == null || checkpointPointers.length == 0 ?
- transactionId : checkpointPointers[checkpointPointers.length - 1];
- doCommit(transactionId, writePointer, edit.getChanges(),
- edit.getCommitPointer(), edit.getCanCommit());
- break;
- case INVALID:
- doInvalidate(edit.getWritePointer());
- break;
- case ABORTED:
- type = edit.getType();
- // Check if transaction edit needs to be migrated to have type. Previous versions of
- // ABORTED edits did not contain type.
- // This can be removed when we stop supporting TransactionEditCodecV2.
- if (type == null) {
- InProgressTx inProgressTx = inProgress.get(edit.getWritePointer());
- if (inProgressTx != null) {
- type = inProgressTx.getType();
- } else {
- // If transaction is not in-progress, then it has either been already aborted or invalidated.
- // We cannot determine the transaction's state based on current information, to be safe invalidate it.
- LOG.warn("Invalidating transaction {} as it's type cannot be determined during replay",
- edit.getWritePointer());
- doInvalidate(edit.getWritePointer());
- break;
- }
- }
- doAbort(edit.getWritePointer(), edit.getCheckpointPointers(), type);
- break;
- case TRUNCATE_INVALID_TX:
- if (edit.getTruncateInvalidTxTime() != 0) {
- doTruncateInvalidTxBefore(edit.getTruncateInvalidTxTime());
- } else {
- doTruncateInvalidTx(edit.getTruncateInvalidTx());
- }
- break;
- case CHECKPOINT:
- doCheckpoint(edit.getWritePointer(), edit.getParentWritePointer());
- break;
- default:
- // unknown type!
- throw new IllegalArgumentException("Invalid state for WAL entry: " + edit.getState());
- }
- }
- } catch (IOException ioe) {
- throw Throwables.propagate(ioe);
- } catch (InvalidTruncateTimeException e) {
- throw Throwables.propagate(e);
- }
- LOG.info("Read " + editCnt + " edits from log " + log.getName());
- }
- }
-
- @Override
- public void doStop() {
- Stopwatch timer = new Stopwatch().start();
- LOG.info("Shutting down gracefully...");
- // signal the cleanup thread to stop
- if (cleanupThread != null) {
- cleanupThread.shutdown();
- try {
- cleanupThread.join(30000L);
- } catch (InterruptedException ie) {
- LOG.warn("Interrupted waiting for cleanup thread to stop");
- Thread.currentThread().interrupt();
- }
- }
- if (metricsThread != null) {
- metricsThread.shutdown();
- try {
- metricsThread.join(30000L);
- } catch (InterruptedException ie) {
- LOG.warn("Interrupted waiting for cleanup thread to stop");
- Thread.currentThread().interrupt();
- }
- }
- if (snapshotThread != null) {
- // this will trigger a final snapshot on stop
- snapshotThread.shutdown();
- try {
- snapshotThread.join(30000L);
- } catch (InterruptedException ie) {
- LOG.warn("Interrupted waiting for snapshot thread to stop");
- Thread.currentThread().interrupt();
- }
- }
-
- persistor.stopAndWait();
- txMetricsCollector.stop();
- timer.stop();
- LOG.info("Took " + timer + " to stop");
- notifyStopped();
- }
-
- /**
- * Immediately shuts down the service, without going through the normal close process.
- * @param message A message describing the source of the failure.
- * @param error Any exception that caused the failure.
- */
- private void abortService(String message, Throwable error) {
- if (isRunning()) {
- LOG.error("Aborting transaction manager due to: " + message, error);
- notifyFailed(error);
- }
- }
-
- private void ensureAvailable() {
- Preconditions.checkState(isRunning(), "Transaction Manager is not running.");
- }
-
- /**
- * Start a short transaction with the default timeout.
- */
- public Transaction startShort() {
- return startShort(defaultTimeout);
- }
-
- /**
- * Start a short transaction with a given timeout.
- * @param timeoutInSeconds the time out period in seconds.
- */
- public Transaction startShort(int timeoutInSeconds) {
- Preconditions.checkArgument(timeoutInSeconds > 0, "timeout must be positive but is %s", timeoutInSeconds);
- txMetricsCollector.rate("start.short");
- Stopwatch timer = new Stopwatch().start();
- long expiration = getTxExpiration(timeoutInSeconds);
- Transaction tx = startTx(expiration, TransactionType.SHORT);
- txMetricsCollector.histogram("start.short.latency", (int) timer.elapsedMillis());
- return tx;
- }
-
- private static long getTxExpiration(long timeoutInSeconds) {
- long currentTime = System.currentTimeMillis();
- return currentTime + TimeUnit.SECONDS.toMillis(timeoutInSeconds);
- }
-
- public static long getTxExpirationFromWritePointer(long writePointer, long timeoutInSeconds) {
- return writePointer / TxConstants.MAX_TX_PER_MS + TimeUnit.SECONDS.toMillis(timeoutInSeconds);
- }
-
- private long getNextWritePointer() {
- // We want to align tx ids with current time. We assume that tx ids are sequential, but not less than
- // System.currentTimeMillis() * MAX_TX_PER_MS.
- return Math.max(lastWritePointer + 1, System.currentTimeMillis() * TxConstants.MAX_TX_PER_MS);
- }
-
- /**
- * Start a long transaction. Long transactions and do not participate in conflict detection. Also, aborting a long
- * transaction moves it to the invalid list because we assume that its writes cannot be rolled back.
- */
- public Transaction startLong() {
- txMetricsCollector.rate("start.long");
- Stopwatch timer = new Stopwatch().start();
- long expiration = getTxExpiration(defaultLongTimeout);
- Transaction tx = startTx(expiration, TransactionType.LONG);
- txMetricsCollector.histogram("start.long.latency", (int) timer.elapsedMillis());
- return tx;
- }
-
- private Transaction startTx(long expiration, TransactionType type) {
- Transaction tx = null;
- long txid;
- // guard against changes to the transaction log while processing
- this.logReadLock.lock();
- try {
- synchronized (this) {
- ensureAvailable();
- txid = getNextWritePointer();
- tx = createTransaction(txid, type);
- addInProgressAndAdvance(tx.getTransactionId(), tx.getVisibilityUpperBound(), expiration, type);
- }
- // appending to WAL out of global lock for concurrent performance
- // we should still be able to arrive at the same state even if log entries are out of order
- appendToLog(TransactionEdit.createStarted(tx.getTransactionId(), tx.getVisibilityUpperBound(), expiration, type));
- } finally {
- this.logReadLock.unlock();
- }
- return tx;
- }
-
- private void addInProgressAndAdvance(long writePointer, long visibilityUpperBound,
- long expiration, TransactionType type) {
- inProgress.put(writePointer, new InProgressTx(visibilityUpperBound, expiration, type));
- advanceWritePointer(writePointer);
- }
-
- private void advanceWritePointer(long writePointer) {
- // don't move the write pointer back if we have out of order transaction log entries
- if (writePointer > lastWritePointer) {
- lastWritePointer = writePointer;
- }
- }
-
- public boolean canCommit(Transaction tx, Collection<byte[]> changeIds) throws TransactionNotInProgressException {
- txMetricsCollector.rate("canCommit");
- Stopwatch timer = new Stopwatch().start();
- if (inProgress.get(tx.getTransactionId()) == null) {
- // invalid transaction, either this has timed out and moved to invalid, or something else is wrong.
- if (invalid.contains(tx.getTransactionId())) {
- throw new TransactionNotInProgressException(
- String.format("canCommit() is called for transaction %d that is not in progress (it is known to be invalid)",
- tx.getTransactionId()));
- } else {
- throw new TransactionNotInProgressException(
- String.format("canCommit() is called for transaction %d that is not in progress", tx.getTransactionId()));
- }
- }
-
- Set<ChangeId> set = Sets.newHashSetWithExpectedSize(changeIds.size());
- for (byte[] change : changeIds) {
- set.add(new ChangeId(change));
- }
-
- if (hasConflicts(tx, set)) {
- return false;
- }
- // guard against changes to the transaction log while processing
- this.logReadLock.lock();
- try {
- synchronized (this) {
- ensureAvailable();
- addCommittingChangeSet(tx.getTransactionId(), set);
- }
- appendToLog(TransactionEdit.createCommitting(tx.getTransactionId(), set));
- } finally {
- this.logReadLock.unlock();
- }
- txMetricsCollector.histogram("canCommit.latency", (int) timer.elapsedMillis());
- return true;
- }
-
- private void addCommittingChangeSet(long writePointer, Set<ChangeId> changes) {
- committingChangeSets.put(writePointer, changes);
- }
-
- public boolean commit(Transaction tx) throws TransactionNotInProgressException {
- txMetricsCollector.rate("commit");
- Stopwatch timer = new Stopwatch().start();
- Set<ChangeId> changeSet = null;
- boolean addToCommitted = true;
- long commitPointer;
- // guard against changes to the transaction log while processing
- this.logReadLock.lock();
- try {
- synchronized (this) {
- ensureAvailable();
- // we record commits at the first not-yet assigned transaction id to simplify clearing out change sets that
- // are no longer visible by any in-progress transactions
- commitPointer = lastWritePointer + 1;
- if (inProgress.get(tx.getTransactionId()) == null) {
- // invalid transaction, either this has timed out and moved to invalid, or something else is wrong.
- if (invalid.contains(tx.getTransactionId())) {
- throw new TransactionNotInProgressException(
- String.format("canCommit() is called for transaction %d that is not in progress " +
- "(it is known to be invalid)", tx.getTransactionId()));
- } else {
- throw new TransactionNotInProgressException(
- String.format("canCommit() is called for transaction %d that is not in progress", tx.getTransactionId()));
- }
- }
-
- // these should be atomic
- // NOTE: whether we succeed or not we don't need to keep changes in committing state: same tx cannot
- // be attempted to commit twice
- changeSet = committingChangeSets.remove(tx.getTransactionId());
-
- if (changeSet != null) {
- // double-checking if there are conflicts: someone may have committed since canCommit check
- if (hasConflicts(tx, changeSet)) {
- return false;
- }
- } else {
- // no changes
- addToCommitted = false;
- }
- doCommit(tx.getTransactionId(), tx.getWritePointer(), changeSet, commitPointer, addToCommitted);
- }
- appendToLog(TransactionEdit.createCommitted(tx.getTransactionId(), changeSet, commitPointer, addToCommitted));
- } finally {
- this.logReadLock.unlock();
- }
- txMetricsCollector.histogram("commit.latency", (int) timer.elapsedMillis());
- return true;
- }
-
- private void doCommit(long transactionId, long writePointer, Set<ChangeId> changes, long commitPointer,
- boolean addToCommitted) {
- // In case this method is called when loading a previous WAL, we need to remove the tx from these sets
- committingChangeSets.remove(transactionId);
- if (addToCommitted && !changes.isEmpty()) {
- // No need to add empty changes to the committed change sets, they will never trigger any conflict
-
- // Record the committed change set with the next writePointer as the commit time.
- // NOTE: we use current next writePointer as key for the map, hence we may have multiple txs changesets to be
- // stored under one key
- Set<ChangeId> changeIds = committedChangeSets.get(commitPointer);
- if (changeIds != null) {
- // NOTE: we modify the new set to prevent concurrent modification exception, as other threads (e.g. in
- // canCommit) use it unguarded
- changes.addAll(changeIds);
- }
- committedChangeSets.put(commitPointer, changes);
- }
- // remove from in-progress set, so that it does not get excluded in the future
- InProgressTx previous = inProgress.remove(transactionId);
- if (previous == null) {
- // tx was not in progress! perhaps it timed out and is invalid? try to remove it there.
- if (invalid.rem(transactionId)) {
- invalidArray = invalid.toLongArray();
- LOG.info("Tx invalid list: removed committed tx {}", transactionId);
- }
- }
- // moving read pointer
- moveReadPointerIfNeeded(writePointer);
-
- // All committed change sets that are smaller than the earliest started transaction can be removed.
- // here we ignore transactions that have no timeout, they are long-running and don't participate in
- // conflict detection.
- // TODO: for efficiency, can we do this once per-log in replayLogs instead of once per edit?
- committedChangeSets.headMap(TxUtils.getFirstShortInProgress(inProgress)).clear();
- }
-
- public void abort(Transaction tx) {
- // guard against changes to the transaction log while processing
- txMetricsCollector.rate("abort");
- Stopwatch timer = new Stopwatch().start();
- this.logReadLock.lock();
- try {
- synchronized (this) {
- ensureAvailable();
- doAbort(tx.getTransactionId(), tx.getCheckpointWritePointers(), tx.getType());
- }
- appendToLog(TransactionEdit.createAborted(tx.getTransactionId(), tx.getType(), tx.getCheckpointWritePointers()));
- txMetricsCollector.histogram("abort.latency", (int) timer.elapsedMillis());
- } finally {
- this.logReadLock.unlock();
- }
- }
-
- private void doAbort(long writePointer, long[] checkpointWritePointers, TransactionType type) {
- committingChangeSets.remove(writePointer);
-
- if (type == TransactionType.LONG) {
- // Long running transactions cannot be aborted as their change sets are not saved,
- // and hence the changes cannot be rolled back. Invalidate the long running transaction instead.
- doInvalidate(writePointer);
- return;
- }
-
- // makes tx visible (assumes that all operations were rolled back)
- // remove from in-progress set, so that it does not get excluded in the future
- InProgressTx removed = inProgress.remove(writePointer);
- if (removed == null) {
- // tx was not in progress! perhaps it timed out and is invalid? try to remove it there.
- if (invalid.rem(writePointer)) {
- // remove any invalidated checkpoint pointers
- // this will only be present if the parent write pointer was also invalidated
- if (checkpointWritePointers != null) {
- for (int i = 0; i < checkpointWritePointers.length; i++) {
- invalid.rem(checkpointWritePointers[i]);
- }
- }
- invalidArray = invalid.toLongArray();
- LOG.info("Tx invalid list: removed aborted tx {}", writePointer);
- // removed a tx from excludes: must move read pointer
- moveReadPointerIfNeeded(writePointer);
- }
- } else {
- // removed a tx from excludes: must move read pointer
- moveReadPointerIfNeeded(writePointer);
- }
- }
-
- public boolean invalidate(long tx) {
- // guard against changes to the transaction log while processing
- txMetricsCollector.rate("invalidate");
- Stopwatch timer = new Stopwatch().start();
- this.logReadLock.lock();
- try {
- boolean success;
- synchronized (this) {
- ensureAvailable();
- success = doInvalidate(tx);
- }
- appendToLog(TransactionEdit.createInvalid(tx));
- txMetricsCollector.histogram("invalidate.latency", (int) timer.elapsedMillis());
- return success;
- } finally {
- this.logReadLock.unlock();
- }
- }
-
- private boolean doInvalidate(long writePointer) {
- Set<ChangeId> previousChangeSet = committingChangeSets.remove(writePointer);
- // remove from in-progress set, so that it does not get excluded in the future
- InProgressTx previous = inProgress.remove(writePointer);
- // This check is to prevent from invalidating committed transactions
- if (previous != null || previousChangeSet != null) {
- // add tx to invalids
- invalid.add(writePointer);
- if (previous == null) {
- LOG.debug("Invalidating tx {} in committing change sets but not in-progress", writePointer);
- } else {
- // invalidate any checkpoint write pointers
- LongArrayList childWritePointers = previous.getCheckpointWritePointers();
- if (childWritePointers != null) {
- for (int i = 0; i < childWritePointers.size(); i++) {
- invalid.add(childWritePointers.get(i));
- }
- }
- }
- LOG.info("Tx invalid list: added tx {} because of invalidate", writePointer);
- // todo: find a more efficient way to keep this sorted. Could it just be an array?
- Collections.sort(invalid);
- invalidArray = invalid.toLongArray();
- if (previous != null && !previous.isLongRunning()) {
- // tx was short-running: must move read pointer
- moveReadPointerIfNeeded(writePointer);
- }
- return true;
- }
- return false;
- }
-
- /**
- * Removes the given transaction ids from the invalid list.
- * @param invalidTxIds transaction ids
- * @return true if invalid list got changed, false otherwise
- */
- public boolean truncateInvalidTx(Set<Long> invalidTxIds) {
- // guard against changes to the transaction log while processing
- txMetricsCollector.rate("truncateInvalidTx");
- Stopwatch timer = new Stopwatch().start();
- this.logReadLock.lock();
- try {
- boolean success;
- synchronized (this) {
- ensureAvailable();
- success = doTruncateInvalidTx(invalidTxIds);
- }
- appendToLog(TransactionEdit.createTruncateInvalidTx(invalidTxIds));
- txMetricsCollector.histogram("truncateInvalidTx.latency", (int) timer.elapsedMillis());
- return success;
- } finally {
- this.logReadLock.unlock();
- }
- }
-
- private boolean doTruncateInvalidTx(Set<Long> invalidTxIds) {
- LOG.info("Removing tx ids {} from invalid list", invalidTxIds);
- boolean success = invalid.removeAll(invalidTxIds);
- if (success) {
- invalidArray = invalid.toLongArray();
- }
- return success;
- }
-
- /**
- * Removes all transaction ids started before the given time from invalid list.
- * @param time time in milliseconds
- * @return true if invalid list got changed, false otherwise
- * @throws InvalidTruncateTimeException if there are any in-progress transactions started before given time
- */
- public boolean truncateInvalidTxBefore(long time) throws InvalidTruncateTimeException {
- // guard against changes to the transaction log while processing
- txMetricsCollector.rate("truncateInvalidTxBefore");
- Stopwatch timer = new Stopwatch().start();
- this.logReadLock.lock();
- try {
- boolean success;
- synchronized (this) {
- ensureAvailable();
- success = doTruncateInvalidTxBefore(time);
- }
- appendToLog(TransactionEdit.createTruncateInvalidTxBefore(time));
- txMetricsCollector.histogram("truncateInvalidTxBefore.latency", (int) timer.elapsedMillis());
- return success;
- } finally {
- this.logReadLock.unlock();
- }
- }
-
- private boolean doTruncateInvalidTxBefore(long time) throws InvalidTruncateTimeException {
- LOG.info("Removing tx ids before {} from invalid list", time);
- long truncateWp = time * TxConstants.MAX_TX_PER_MS;
- // Check if there any in-progress transactions started earlier than truncate time
- if (inProgress.lowerKey(truncateWp) != null) {
- throw new InvalidTruncateTimeException("Transactions started earlier than " + time + " are in-progress");
- }
-
- // Find all invalid transactions earlier than truncateWp
- Set<Long> toTruncate = Sets.newHashSet();
- for (long wp : invalid) {
- // invalid list is sorted, hence can stop as soon as we reach a wp >= truncateWp
- if (wp >= truncateWp) {
- break;
- }
- toTruncate.add(wp);
- }
- return doTruncateInvalidTx(toTruncate);
- }
-
- public Transaction checkpoint(Transaction originalTx) throws TransactionNotInProgressException {
- txMetricsCollector.rate("checkpoint");
- Stopwatch timer = new Stopwatch().start();
-
- Transaction checkpointedTx = null;
- long txId = originalTx.getTransactionId();
- long newWritePointer = 0;
- // guard against changes to the transaction log while processing
- this.logReadLock.lock();
- try {
- synchronized (this) {
- ensureAvailable();
- // check that the parent tx is in progress
- InProgressTx parentTx = inProgress.get(txId);
- if (parentTx == null) {
- if (invalid.contains(txId)) {
- throw new TransactionNotInProgressException(
- String.format("Transaction %d is not in progress because it was invalidated", txId));
- } else {
- throw new TransactionNotInProgressException(
- String.format("Transaction %d is not in progress", txId));
- }
- }
- newWritePointer = getNextWritePointer();
- doCheckpoint(newWritePointer, txId);
- // create a new transaction with the same read snapshot, plus the additional checkpoint write pointer
- // the same read snapshot is maintained to
- checkpointedTx = new Transaction(originalTx, newWritePointer,
- parentTx.getCheckpointWritePointers().toLongArray());
- }
- // appending to WAL out of global lock for concurrent performance
- // we should still be able to arrive at the same state even if log entries are out of order
- appendToLog(TransactionEdit.createCheckpoint(newWritePointer, txId));
- } finally {
- this.logReadLock.unlock();
- }
- txMetricsCollector.histogram("checkpoint.latency", (int) timer.elapsedMillis());
-
- return checkpointedTx;
- }
-
- private void doCheckpoint(long newWritePointer, long parentWritePointer) {
- InProgressTx existingTx = inProgress.get(parentWritePointer);
- existingTx.addCheckpointWritePointer(newWritePointer);
- advanceWritePointer(newWritePointer);
- }
-
- // hack for exposing important metric
- public int getExcludedListSize() {
- return invalid.size() + inProgress.size();
- }
-
- /**
- * @return the size of invalid list
- */
- public int getInvalidSize() {
- return this.invalid.size();
- }
-
- int getCommittedSize() {
- return this.committedChangeSets.size();
- }
-
- private boolean hasConflicts(Transaction tx, Set<ChangeId> changeIds) {
- if (changeIds.isEmpty()) {
- return false;
- }
-
- for (Map.Entry<Long, Set<ChangeId>> changeSet : committedChangeSets.entrySet()) {
- // If commit time is greater than tx read-pointer,
- // basically not visible but committed means "tx committed after given tx was started"
- if (changeSet.getKey() > tx.getTransactionId()) {
- if (overlap(changeSet.getValue(), changeIds)) {
- return true;
- }
- }
- }
- return false;
- }
-
- private boolean overlap(Set<ChangeId> a, Set<ChangeId> b) {
- // iterate over the smaller set, and check for every element in the other set
- if (a.size() > b.size()) {
- for (ChangeId change : b) {
- if (a.contains(change)) {
- return true;
- }
- }
- } else {
- for (ChangeId change : a) {
- if (b.contains(change)) {
- return true;
- }
- }
- }
- return false;
- }
-
- private void moveReadPointerIfNeeded(long committedWritePointer) {
- if (committedWritePointer > readPointer) {
- readPointer = committedWritePointer;
- }
- }
-
- /**
- * Creates a new Transaction. This method only get called from start transaction, which is already
- * synchronized.
- */
- private Transaction createTransaction(long writePointer, TransactionType type) {
- // For holding the first in progress short transaction Id (with timeout >= 0).
- long firstShortTx = Transaction.NO_TX_IN_PROGRESS;
- LongArrayList inProgressIds = new LongArrayList(inProgress.size());
- for (Map.Entry<Long, InProgressTx> entry : inProgress.entrySet()) {
- long txId = entry.getKey();
- inProgressIds.add(txId);
- // add any checkpointed write pointers to the in-progress list
- LongArrayList childIds = entry.getValue().getCheckpointWritePointers();
- if (childIds != null) {
- for (int i = 0; i < childIds.size(); i++) {
- inProgressIds.add(childIds.get(i));
- }
- }
- if (firstShortTx == Transaction.NO_TX_IN_PROGRESS && !entry.getValue().isLongRunning()) {
- firstShortTx = txId;
- }
- }
-
- return new Transaction(readPointer, writePointer, invalidArray, inProgressIds.toLongArray(), firstShortTx, type);
- }
-
- private void appendToLog(TransactionEdit edit) {
- try {
- Stopwatch timer = new Stopwatch().start();
- currentLog.append(edit);
- txMetricsCollector.rate("wal.append.count");
- txMetricsCollector.histogram("wal.append.latency", (int) timer.elapsedMillis());
- } catch (IOException ioe) {
- abortService("Error appending to transaction log", ioe);
- }
- }
-
- private void appendToLog(List<TransactionEdit> edits) {
- try {
- Stopwatch timer = new Stopwatch().start();
- currentLog.append(edits);
- txMetricsCollector.rate("wal.append.count", edits.size());
- txMetricsCollector.histogram("wal.append.latency", (int) timer.elapsedMillis());
- } catch (IOException ioe) {
- abortService("Error appending to transaction log", ioe);
- }
- }
-
- /**
- * Called from the tx service every 10 seconds.
- * This hack is needed because current metrics system is not flexible when it comes to adding new metrics.
- */
- public void logStatistics() {
- LOG.info("Transaction Statistics: write pointer = " + lastWritePointer +
- ", invalid = " + invalid.size() +
- ", in progress = " + inProgress.size() +
- ", committing = " + committingChangeSets.size() +
- ", committed = " + committedChangeSets.size());
- }
-
- private abstract static class DaemonThreadExecutor extends Thread {
- private AtomicBoolean stopped = new AtomicBoolean(false);
-
- public DaemonThreadExecutor(String name) {
- super(name);
- setDaemon(true);
- }
-
- public void run() {
- try {
- while (!isInterrupted() && !stopped.get()) {
- doRun();
- synchronized (stopped) {
- stopped.wait(getSleepMillis());
- }
- }
- } catch (InterruptedException ie) {
- LOG.info("Interrupted thread " + getName());
- }
- // perform any final cleanup
- onShutdown();
- LOG.info("Exiting thread " + getName());
- }
-
- public abstract void doRun();
-
- protected abstract long getSleepMillis();
-
- protected void onShutdown() {
- }
-
- public void shutdown() {
- if (stopped.compareAndSet(false, true)) {
- synchronized (stopped) {
- stopped.notifyAll();
- }
- }
- }
- }
-
- /**
- * Represents some of the info on in-progress tx
- */
- public static final class InProgressTx {
- /** the oldest in progress tx at the time of this tx start */
- private final long visibilityUpperBound;
- private final long expiration;
- private final TransactionType type;
- private LongArrayList checkpointWritePointers = new LongArrayList();
-
- public InProgressTx(long visibilityUpperBound, long expiration, TransactionType type) {
- this(visibilityUpperBound, expiration, type, new LongArrayList());
- }
-
- public InProgressTx(long visibilityUpperBound, long expiration, TransactionType type,
- LongArrayList checkpointWritePointers) {
- this.visibilityUpperBound = visibilityUpperBound;
- this.expiration = expiration;
- this.type = type;
- this.checkpointWritePointers = checkpointWritePointers;
- }
-
- // For backwards compatibility when long running txns were represented with -1 expiration
- @Deprecated
- public InProgressTx(long visibilityUpperBound, long expiration) {
- this(visibilityUpperBound, expiration, null);
- }
-
- public long getVisibilityUpperBound() {
- return visibilityUpperBound;
- }
-
- public long getExpiration() {
- return expiration;
- }
-
- @Nullable
- public TransactionType getType() {
- return type;
- }
-
- public boolean isLongRunning() {
- if (type == null) {
- // for backwards compatibility when long running txns were represented with -1 expiration
- return expiration == -1;
- }
- return type == TransactionType.LONG;
- }
-
- public void addCheckpointWritePointer(long checkpointWritePointer) {
- checkpointWritePointers.add(checkpointWritePointer);
- }
-
- public LongArrayList getCheckpointWritePointers() {
- return checkpointWritePointers;
- }
-
- @Override
- public boolean equals(Object o) {
- if (o == null || !(o instanceof InProgressTx)) {
- return false;
- }
-
- if (this == o) {
- return true;
- }
-
- InProgressTx other = (InProgressTx) o;
- return Objects.equal(visibilityUpperBound, other.getVisibilityUpperBound()) &&
- Objects.equal(expiration, other.getExpiration()) &&
- Objects.equal(type, other.type) &&
- Objects.equal(checkpointWritePointers, other.checkpointWritePointers);
- }
-
- @Override
- public int hashCode() {
- return Objects.hashCode(visibilityUpperBound, expiration, type, checkpointWritePointers);
- }
-
- @Override
- public String toString() {
- return Objects.toStringHelper(this)
- .add("visibilityUpperBound", visibilityUpperBound)
- .add("expiration", expiration)
- .add("type", type)
- .add("checkpointWritePointers", checkpointWritePointers)
- .toString();
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-tephra/blob/9c693743/tephra-core/src/main/java/co/cask/tephra/TransactionServiceMain.java
----------------------------------------------------------------------
diff --git a/tephra-core/src/main/java/co/cask/tephra/TransactionServiceMain.java b/tephra-core/src/main/java/co/cask/tephra/TransactionServiceMain.java
deleted file mode 100644
index 8437017..0000000
--- a/tephra-core/src/main/java/co/cask/tephra/TransactionServiceMain.java
+++ /dev/null
@@ -1,149 +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 co.cask.tephra;
-
-import co.cask.tephra.distributed.TransactionService;
-import co.cask.tephra.runtime.ConfigModule;
-import co.cask.tephra.runtime.DiscoveryModules;
-import co.cask.tephra.runtime.TransactionClientModule;
-import co.cask.tephra.runtime.TransactionModules;
-import co.cask.tephra.runtime.ZKModule;
-import co.cask.tephra.util.ConfigurationFactory;
-import com.google.inject.Guice;
-import com.google.inject.Injector;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.twill.zookeeper.ZKClientService;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.concurrent.CountDownLatch;
-
-/**
- * Driver class to start and stop tx in distributed mode.
- */
-public class TransactionServiceMain {
-
- private static final Logger LOG = LoggerFactory.getLogger(TransactionServiceMain.class);
-
- private Configuration conf;
- private TransactionService txService;
-
- public static void main(String args[]) throws Exception {
- TransactionServiceMain instance = new TransactionServiceMain();
- instance.doMain(args);
- }
-
- public TransactionServiceMain() {
- this(null);
- }
-
- public TransactionServiceMain(Configuration conf) {
- this.conf = conf;
- }
-
- /**
- * The main method. It simply call methods in the same sequence
- * as if the program is started by jsvc.
- */
- public void doMain(final String[] args) throws Exception {
- final CountDownLatch shutdownLatch = new CountDownLatch(1);
- Runtime.getRuntime().addShutdownHook(new Thread() {
- @Override
- public void run() {
- try {
- try {
- TransactionServiceMain.this.stop();
- } finally {
- try {
- TransactionServiceMain.this.destroy();
- } finally {
- shutdownLatch.countDown();
- }
- }
- } catch (Throwable t) {
- LOG.error("Exception when shutting down: " + t.getMessage(), t);
- }
- }
- });
- init(args);
- start();
-
- shutdownLatch.await();
- }
-
- /**
- * Invoked by jsvc to initialize the program.
- */
- public void init(String[] args) {
- if (conf == null) {
- conf = new ConfigurationFactory().get();
- }
- }
-
- /**
- * Invoked by jsvc to start the program.
- */
- public void start() throws Exception {
- Injector injector = Guice.createInjector(
- new ConfigModule(conf),
- new ZKModule(),
- new DiscoveryModules().getDistributedModules(),
- new TransactionModules().getDistributedModules(),
- new TransactionClientModule()
- );
-
- ZKClientService zkClientService = injector.getInstance(ZKClientService.class);
- zkClientService.startAndWait();
-
- // start a tx server
- txService = injector.getInstance(TransactionService.class);
- try {
- LOG.info("Starting {}", getClass().getSimpleName());
- txService.startAndWait();
- } catch (Exception e) {
- System.err.println("Failed to start service: " + e.getMessage());
- }
- }
-
- /**
- * Invoked by jsvc to stop the program.
- */
- public void stop() {
- LOG.info("Stopping {}", getClass().getSimpleName());
- if (txService == null) {
- return;
- }
- try {
- if (txService.isRunning()) {
- txService.stopAndWait();
- }
- } catch (Throwable e) {
- LOG.error("Failed to shutdown transaction service.", e);
- // because shutdown hooks execute concurrently, the logger may be closed already: thus also print it.
- System.err.println("Failed to shutdown transaction service: " + e.getMessage());
- e.printStackTrace(System.err);
- }
- }
-
- /**
- * Invoked by jsvc for resource cleanup.
- */
- public void destroy() { }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-tephra/blob/9c693743/tephra-core/src/main/java/co/cask/tephra/TransactionSystemClient.java
----------------------------------------------------------------------
diff --git a/tephra-core/src/main/java/co/cask/tephra/TransactionSystemClient.java b/tephra-core/src/main/java/co/cask/tephra/TransactionSystemClient.java
deleted file mode 100644
index ca68143..0000000
--- a/tephra-core/src/main/java/co/cask/tephra/TransactionSystemClient.java
+++ /dev/null
@@ -1,142 +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 co.cask.tephra;
-
-import java.io.InputStream;
-import java.util.Collection;
-import java.util.Set;
-
-/**
- * Client talking to transaction system.
- * See also {@link co.cask.tephra.TransactionAware}.
- * todo: explain Omid.
- */
-public interface TransactionSystemClient {
- /**
- * Starts new short transaction.
- * @return instance of {@link co.cask.tephra.Transaction}
- */
- // TODO: "short" and "long" are very misleading names. Use transaction attributes like "detect conflicts or not", etc.
- Transaction startShort();
-
- /**
- * Starts new short transaction.
- * @param timeout the timeout for the transaction
- * @return instance of {@link Transaction}
- */
- Transaction startShort(int timeout);
-
- /**
- * Starts new long transaction.
- * @return instance of {@link Transaction}
- */
- Transaction startLong();
-
- // this pre-commit detects conflicts with other transactions committed so far
- // NOTE: the changes set should not change after this operation, this may help us do some extra optimizations
- // NOTE: there should be time constraint on how long does it take to commit changes by the client after this operation
- // is submitted so that we can cleanup related resources
- // NOTE: as of now you can call this method multiple times, each time the changeSet of tx will be updated. Not sure
- // if we can call it a feature or a side-affect of implementation. It makes more sense to append changeset, but
- // before we really need it we don't do it because it will slow down tx manager throughput.
-
- /**
- * Checks if transaction with the set of changes can be committed. E.g. it can check conflicts with other changes and
- * refuse commit if there are conflicts. It is assumed that no other changes will be done in between this method call
- * and {@link #commit(Transaction)} which may check conflicts again to avoid races.
- * <p/>
- * Since we do conflict detection at commit time as well, this may seem redundant. The idea is to check for conflicts
- * before we persist changes to avoid rollback in case of conflicts as much as possible.
- * NOTE: in some situations we may want to skip this step to save on RPC with a risk of many rollback ops. So by
- * default we take safe path.
- *
- * @param tx transaction to verify
- * @param changeIds ids of changes made by transaction
- * @return true if transaction can be committed otherwise false
- */
- boolean canCommit(Transaction tx, Collection<byte[]> changeIds) throws TransactionNotInProgressException;
-
- /**
- * Makes transaction visible. It will again check conflicts of changes submitted previously with
- * {@link #canCommit(Transaction, java.util.Collection)}
- * @param tx transaction to make visible.
- * @return true if transaction can be committed otherwise false
- */
- boolean commit(Transaction tx) throws TransactionNotInProgressException;
-
- /**
- * Makes transaction visible. You should call it only when all changes of this tx are undone.
- * NOTE: it will not throw {@link TransactionNotInProgressException} if transaction has timed out.
- * @param tx transaction to make visible.
- */
- void abort(Transaction tx);
-
- /**
- * Makes transaction invalid. You should call it if not all changes of this tx could be undone.
- * NOTE: it will not throw {@link TransactionNotInProgressException} if transaction has timed out.
- * @param tx transaction id to invalidate.
- * @return true if transaction has been successfully invalidated
- */
- boolean invalidate(long tx);
-
- /**
- * Performs a checkpoint operation on the current transaction, returning a new Transaction instance with the
- * updated state. A checkpoint operation assigns a new write pointer for the current transaction.
- * @param tx the current transaction to checkpoint
- * @return an updated transaction instance with the new write pointer
- */
- Transaction checkpoint(Transaction tx) throws TransactionNotInProgressException;
-
- /**
- * Retrieves the state of the transaction manager and send it as a stream. The snapshot will not be persisted.
- * @return an input stream containing an encoded snapshot of the transaction manager
- */
- InputStream getSnapshotInputStream() throws TransactionCouldNotTakeSnapshotException;
-
- /**
- * Return the status of the transaction Manager
- * @return a String which denotes the status of txManager
- */
- String status();
-
- /**
- * Resets the state of the transaction manager.
- */
- void resetState();
-
- /**
- * Removes the given transaction ids from the invalid list.
- * @param invalidTxIds transaction ids
- * @return true if invalid list got changed, false otherwise
- */
- boolean truncateInvalidTx(Set<Long> invalidTxIds);
-
- /**
- * Removes all transaction ids started before the given time from invalid list.
- * @param time time in milliseconds
- * @return true if invalid list got changed, false otherwise
- * @throws InvalidTruncateTimeException if there are any in-progress transactions started before given time
- */
- boolean truncateInvalidTxBefore(long time) throws InvalidTruncateTimeException;
-
- /**
- * @return the size of invalid list
- */
- int getInvalidSize();
-}
http://git-wip-us.apache.org/repos/asf/incubator-tephra/blob/9c693743/tephra-core/src/main/java/co/cask/tephra/TxConstants.java
----------------------------------------------------------------------
diff --git a/tephra-core/src/main/java/co/cask/tephra/TxConstants.java b/tephra-core/src/main/java/co/cask/tephra/TxConstants.java
deleted file mode 100644
index 7c96c85..0000000
--- a/tephra-core/src/main/java/co/cask/tephra/TxConstants.java
+++ /dev/null
@@ -1,343 +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 co.cask.tephra;
-
-import co.cask.tephra.snapshot.DefaultSnapshotCodec;
-import co.cask.tephra.snapshot.SnapshotCodecV2;
-import co.cask.tephra.snapshot.SnapshotCodecV3;
-import co.cask.tephra.snapshot.SnapshotCodecV4;
-
-import java.util.concurrent.TimeUnit;
-
-/**
- * Transaction system constants
- */
-public class TxConstants {
- /**
- * Defines what level of conflict detection should be used for transactions. {@code ROW} means that only the
- * table name and row key for each change will be used to determine if the transaction change sets conflict.
- * {@code COLUMN} means that the table name, row key, column family, and column qualifier will all be used to
- * identify write conflicts. {@code NONE} means that no conflict detection will be performed, but transaction
- * clients will still track the current transaction's change set to rollback any persisted changes in the event of
- * a failure. This should only be used where writes to the same coordinate should never conflict, such as
- * append-only data. The default value used by {@code TransactionAwareHTable} implementations is {@code COLUMN}.
- *
- * <p>
- * <strong>Note: for a given table, all clients must use the same conflict detection setting!</strong>
- * Otherwise conflicts will not be flagged correctly.
- * </p>
- */
- public enum ConflictDetection {
- ROW,
- COLUMN,
- NONE
- }
-
- /**
- * Property set for {@code org.apache.hadoop.hbase.HColumnDescriptor} to configure time-to-live on data within
- * the column family. The value given is in milliseconds. Once a cell's data has surpassed the given value in age,
- * the cell's data will no longer be visible and may be garbage collected.
- */
- public static final String PROPERTY_TTL = "dataset.table.ttl";
-
- /**
- * This is how many tx we allow per millisecond, if you care about the system for 100 years:
- * Long.MAX_VALUE / (System.currentTimeMillis() + TimeUnit.DAYS.toMillis(365 * 100)) =
- * (as of Feb 20, 2014) 2,028,653. It is safe and convenient to use 1,000,000 as multiplier:
- * <ul>
- * <li>
- * we hardly can do more than 1 billion txs per second
- * </li>
- * <li>
- * long value will not overflow for 200 years
- * </li>
- * <li>
- * makes reading & debugging easier if multiplier is 10^n
- * </li>
- * </ul>
- */
- public static final long MAX_TX_PER_MS = 1000000;
-
- /**
- * Since HBase {@code Delete} operations cannot be undone at the same timestamp, "deleted" data is instead
- * overwritten with an empty {@code byte[]} to flag it as removed. Cells with empty values will be filtered out
- * of the results for read operations. If cells with empty values should be included in results (meaning data
- * cannot be transactionally deleted), then set this configuration property to true.
- */
- public static final String ALLOW_EMPTY_VALUES_KEY = "data.tx.allow.empty.values";
- public static final boolean ALLOW_EMPTY_VALUES_DEFAULT = false;
-
- /**
- * Key used to set the serialized transaction as an attribute on Get and Scan operations.
- */
- public static final String TX_OPERATION_ATTRIBUTE_KEY = "cask.tx";
- /**
- * Key used to flag a delete operation as part of a transaction rollback. This is used so that the
- * {@code TransactionProcessor} coprocessor loaded on a table can differentiate between deletes issued
- * as part of a normal client operation versus those performed when rolling back a transaction.
- */
- public static final String TX_ROLLBACK_ATTRIBUTE_KEY = "cask.tx.rollback";
-
- /**
- * Column qualifier used for a special delete marker tombstone, which identifies an entire column family as deleted.
- */
- public static final byte[] FAMILY_DELETE_QUALIFIER = new byte[0];
-
- // Constants for monitoring status
- public static final String STATUS_OK = "OK";
- public static final String STATUS_NOTOK = "NOTOK";
-
- /**
- * Indicates whether data written before Tephra was enabled on a table should be readable. Reading non-transactional
- * data can lead to slight performance penalty. Hence it is disabled by default.
- * @see <a href="https://issues.cask.co/browse/TEPHRA-89">TEPHRA-89</a>
- */
- public static final String READ_NON_TX_DATA = "data.tx.read.pre.existing";
- public static final boolean DEFAULT_READ_NON_TX_DATA = false;
-
- /**
- * TransactionManager configuration.
- */
- public static final class Manager {
- // TransactionManager configuration
- public static final String CFG_DO_PERSIST = "tx.persist";
- /** Directory in HDFS used for transaction snapshot and log storage. */
- public static final String CFG_TX_SNAPSHOT_DIR = "data.tx.snapshot.dir";
- /** Directory on the local filesystem used for transaction snapshot and log storage. */
- public static final String CFG_TX_SNAPSHOT_LOCAL_DIR = "data.tx.snapshot.local.dir";
- /** How often to clean up timed out transactions, in seconds, or 0 for no cleanup. */
- public static final String CFG_TX_CLEANUP_INTERVAL = "data.tx.cleanup.interval";
- /** The user id to access HDFS if not running in secure HDFS. */
- public static final String CFG_TX_HDFS_USER = "data.tx.hdfs.user";
- /** Default value for how often to check in-progress transactions for expiration, in seconds. */
- public static final int DEFAULT_TX_CLEANUP_INTERVAL = 10;
- /**
- * The timeout for a transaction, in seconds. If the transaction is not finished in that time,
- * it is marked invalid.
- */
- public static final String CFG_TX_TIMEOUT = "data.tx.timeout";
- /** Default value for transaction timeout, in seconds. */
- public static final int DEFAULT_TX_TIMEOUT = 30;
- /**
- * The timeout for a long running transaction, in seconds. If the transaction is not finished in that time,
- * it is marked invalid.
- */
- public static final String CFG_TX_LONG_TIMEOUT = "data.tx.long.timeout";
- /** Default value for long running transaction timeout, in seconds. */
- public static final int DEFAULT_TX_LONG_TIMEOUT = (int) TimeUnit.DAYS.toSeconds(1);
- /** The frequency (in seconds) to perform periodic snapshots, or 0 for no periodic snapshots. */
- public static final String CFG_TX_SNAPSHOT_INTERVAL = "data.tx.snapshot.interval";
- /** Default value for frequency of periodic snapshots of transaction state. */
- public static final long DEFAULT_TX_SNAPSHOT_INTERVAL = 300;
- /** Number of most recent transaction snapshots to retain. */
- public static final String CFG_TX_SNAPSHOT_RETAIN = "data.tx.snapshot.retain";
- /** Default value for number of most recent snapshots to retain. */
- public static final int DEFAULT_TX_SNAPSHOT_RETAIN = 10;
- }
-
- /**
- * TransactionService configuration.
- */
- public static final class Service {
-
- /** for the zookeeper quorum string for leader election for tx server. */
- public static final String CFG_DATA_TX_ZOOKEEPER_QUORUM
- = "data.tx.zookeeper.quorum";
-
- /** for the name used to announce service availability to discovery service */
- public static final String CFG_DATA_TX_DISCOVERY_SERVICE_NAME
- = "data.tx.discovery.service.name";
-
- /** for the port of the tx server. */
- public static final String CFG_DATA_TX_BIND_PORT
- = "data.tx.bind.port";
-
- /** for the address (hostname) of the tx server. */
- public static final String CFG_DATA_TX_BIND_ADDRESS
- = "data.tx.bind.address";
-
- /** the number of IO threads in the tx service. */
- public static final String CFG_DATA_TX_SERVER_IO_THREADS
- = "data.tx.server.io.threads";
-
- /** the number of handler threads in the tx service. */
- public static final String CFG_DATA_TX_SERVER_THREADS
- = "data.tx.server.threads";
-
- public static final String CFG_DATA_TX_THRIFT_MAX_READ_BUFFER
- = "data.tx.thrift.max.read.buffer";
-
- public static final String DEFAULT_DATA_TX_DISCOVERY_SERVICE_NAME
- = "transaction";
-
- /** default tx service port. */
- public static final int DEFAULT_DATA_TX_BIND_PORT
- = 15165;
-
- /** default tx service address. */
- public static final String DEFAULT_DATA_TX_BIND_ADDRESS
- = "0.0.0.0";
-
- /** default number of handler IO threads in tx service. */
- public static final int DEFAULT_DATA_TX_SERVER_IO_THREADS
- = 2;
-
- /** default number of handler threads in tx service. */
- public static final int DEFAULT_DATA_TX_SERVER_THREADS
- = 20;
-
- /** default thrift max read buffer size */
- public static final int DEFAULT_DATA_TX_THRIFT_MAX_READ_BUFFER
- = 16 * 1024 * 1024;
-
- // Configuration key names and defaults used by tx client.
-
- /** to specify the tx client socket timeout in ms. */
- public static final String CFG_DATA_TX_CLIENT_TIMEOUT
- = "data.tx.client.timeout";
-
- /** to specify the tx client provider strategy. */
- public static final String CFG_DATA_TX_CLIENT_PROVIDER
- = "data.tx.client.provider";
-
- /** to specify the number of threads for client provider "pool". */
- public static final String CFG_DATA_TX_CLIENT_COUNT
- = "data.tx.client.count";
-
- /** timeout (in milliseconds) for obtaining client from client provider "pool". */
- public static final String CFG_DATA_TX_CLIENT_OBTAIN_TIMEOUT_MS
- = "data.tx.client.obtain.timeout";
-
- /** to specify the retry strategy for a failed thrift call. */
- public static final String CFG_DATA_TX_CLIENT_RETRY_STRATEGY
- = "data.tx.client.retry.strategy";
-
- /** to specify the number of times to retry a failed thrift call. */
- public static final String CFG_DATA_TX_CLIENT_ATTEMPTS
- = "data.tx.client.retry.attempts";
-
- /** to specify the initial sleep time for retry strategy backoff. */
- public static final String CFG_DATA_TX_CLIENT_BACKOFF_INITIAL
- = "data.tx.client.retry.backoff.initial";
-
- /** to specify the backoff factor for retry strategy backoff. */
- public static final String CFG_DATA_TX_CLIENT_BACKOFF_FACTOR
- = "data.tx.client.retry.backoff.factor";
-
- /** to specify the sleep time limit for retry strategy backoff. */
- public static final String CFG_DATA_TX_CLIENT_BACKOFF_LIMIT
- = "data.tx.client.retry.backoff.limit";
-
- /** the default tx client socket timeout in milli seconds. */
- public static final int DEFAULT_DATA_TX_CLIENT_TIMEOUT_MS
- = 30 * 1000;
-
- /** default number of tx clients for client provider "pool". */
- public static final int DEFAULT_DATA_TX_CLIENT_COUNT
- = 50;
-
- /** default timeout (in milliseconds) for obtaining client from client provider "pool". */
- public static final long DEFAULT_DATA_TX_CLIENT_OBTAIN_TIMEOUT_MS
- = TimeUnit.SECONDS.toMillis(3);
-
- /** default tx client provider strategy. */
- public static final String DEFAULT_DATA_TX_CLIENT_PROVIDER
- = "pool";
-
- /** retry strategy for thrift clients, e.g. backoff, or n-times. */
- public static final String DEFAULT_DATA_TX_CLIENT_RETRY_STRATEGY
- = "backoff";
-
- /** default number of attempts for strategy n-times. */
- public static final int DEFAULT_DATA_TX_CLIENT_ATTEMPTS
- = 2;
-
- /** default initial sleep is 100ms. */
- public static final int DEFAULT_DATA_TX_CLIENT_BACKOFF_INITIAL
- = 100;
-
- /** default backoff factor is 4. */
- public static final int DEFAULT_DATA_TX_CLIENT_BACKOFF_FACTOR
- = 4;
-
- /** default sleep limit is 30 sec. */
- public static final int DEFAULT_DATA_TX_CLIENT_BACKOFF_LIMIT
- = 30 * 1000;
- }
-
- /**
- * Configuration properties for metrics reporting
- */
- public static final class Metrics {
- /**
- * Frequency at which metrics should be reported, in seconds.
- */
- public static final String REPORT_PERIOD_KEY = "data.tx.metrics.period";
- /**
- * Default report period for metrics, in seconds.
- */
- public static final int REPORT_PERIOD_DEFAULT = 60;
- }
-
- /**
- * Configuration properties used by HBase
- */
- public static final class HBase {
- public static final String ZOOKEEPER_QUORUM = "hbase.zookeeper.quorum";
- public static final String ZK_SESSION_TIMEOUT = "zookeeper.session.timeout";
- public static final int DEFAULT_ZK_SESSION_TIMEOUT = 180 * 1000;
- }
-
- /**
- * Configuration for the TransactionDataJanitor coprocessor.
- */
- public static final class DataJanitor {
- /**
- * Whether or not the TransactionDataJanitor coprocessor should be enabled on tables.
- * Disable for testing.
- */
- public static final String CFG_TX_JANITOR_ENABLE = "data.tx.janitor.enable";
- public static final boolean DEFAULT_TX_JANITOR_ENABLE = true;
- }
-
- /**
- * Configuration for the transaction snapshot persistence.
- */
- public static final class Persist {
- /**
- * The class names of all known transaction snapshot codecs.
- */
- public static final String CFG_TX_SNAPHOT_CODEC_CLASSES = "data.tx.snapshot.codecs";
- public static final Class[] DEFAULT_TX_SNAPHOT_CODEC_CLASSES =
- { DefaultSnapshotCodec.class, SnapshotCodecV2.class, SnapshotCodecV3.class, SnapshotCodecV4.class };
- }
-
- /**
- * Configuration for transaction log edit entries
- */
- public static final class TransactionLog {
- /**
- * Key used to denote the number of entries appended.
- */
- public static final String NUM_ENTRIES_APPENDED = "count";
- public static final String VERSION_KEY = "version";
- public static final byte CURRENT_VERSION = 2;
- }
-
-}