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:43 UTC
[26/52] [abbrv] incubator-omid git commit: Move com.yahoo ->
org.apache
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/timestamp-storage/src/main/java/org/apache/omid/timestamp/storage/ZKTimestampStorage.java
----------------------------------------------------------------------
diff --git a/timestamp-storage/src/main/java/org/apache/omid/timestamp/storage/ZKTimestampStorage.java b/timestamp-storage/src/main/java/org/apache/omid/timestamp/storage/ZKTimestampStorage.java
new file mode 100644
index 0000000..02bd5f6
--- /dev/null
+++ b/timestamp-storage/src/main/java/org/apache/omid/timestamp/storage/ZKTimestampStorage.java
@@ -0,0 +1,90 @@
+/**
+ * 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.timestamp.storage;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.recipes.atomic.AtomicValue;
+import org.apache.curator.framework.recipes.atomic.DistributedAtomicLong;
+import org.apache.curator.retry.RetryNTimes;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.inject.Inject;
+import java.io.IOException;
+
+import static org.apache.omid.timestamp.storage.ZKTimestampPaths.TIMESTAMP_ZNODE;
+
+class ZKTimestampStorage implements TimestampStorage {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ZKTimestampStorage.class);
+
+ static final long INITIAL_MAX_TS_VALUE = 0;
+
+ private final DistributedAtomicLong timestamp;
+
+ @Inject
+ public ZKTimestampStorage(CuratorFramework zkClient) throws Exception {
+ LOG.info("ZK Client state {}", zkClient.getState());
+ timestamp = new DistributedAtomicLong(zkClient, TIMESTAMP_ZNODE, new RetryNTimes(3, 1000)); // TODO Configure
+ // this?
+ if (timestamp.initialize(INITIAL_MAX_TS_VALUE)) {
+ LOG.info("Timestamp value in ZNode initialized to {}", INITIAL_MAX_TS_VALUE);
+ }
+ }
+
+ @Override
+ public void updateMaxTimestamp(long previousMaxTimestamp, long newMaxTimestamp) throws IOException {
+
+ if (newMaxTimestamp < 0) {
+ LOG.error("Negative value received for maxTimestamp: {}", newMaxTimestamp);
+ throw new IllegalArgumentException();
+ }
+ if (newMaxTimestamp <= previousMaxTimestamp) {
+ LOG.error("maxTimestamp {} <= previousMaxTimesamp: {}", newMaxTimestamp, previousMaxTimestamp);
+ throw new IllegalArgumentException();
+ }
+ AtomicValue<Long> compareAndSet;
+ try {
+ compareAndSet = timestamp.compareAndSet(previousMaxTimestamp, newMaxTimestamp);
+ } catch (Exception e) {
+ throw new IOException("Problem setting timestamp in ZK", e);
+ }
+ if (!compareAndSet.succeeded()) { // We have to explicitly check for success (See Curator doc)
+ throw new IOException("GetAndSet operation for storing timestamp in ZK did not succeed "
+ + compareAndSet.preValue() + " " + compareAndSet.postValue());
+ }
+
+ }
+
+ @Override
+ public long getMaxTimestamp() throws IOException {
+
+ AtomicValue<Long> atomicValue;
+ try {
+ atomicValue = timestamp.get();
+ } catch (Exception e) {
+ throw new IOException("Problem getting data from ZK", e);
+ }
+ if (!atomicValue.succeeded()) { // We have to explicitly check for success (See Curator doc)
+ throw new IOException("Get operation to obtain timestamp from ZK did not succeed");
+ }
+ return atomicValue.postValue();
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/timestamp-storage/src/main/java/org/apache/omid/timestamp/storage/ZKTimestampStorageModule.java
----------------------------------------------------------------------
diff --git a/timestamp-storage/src/main/java/org/apache/omid/timestamp/storage/ZKTimestampStorageModule.java b/timestamp-storage/src/main/java/org/apache/omid/timestamp/storage/ZKTimestampStorageModule.java
new file mode 100644
index 0000000..48e46c5
--- /dev/null
+++ b/timestamp-storage/src/main/java/org/apache/omid/timestamp/storage/ZKTimestampStorageModule.java
@@ -0,0 +1,37 @@
+/**
+ * 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.timestamp.storage;
+
+import com.google.inject.AbstractModule;
+import org.apache.curator.framework.CuratorFramework;
+
+import javax.inject.Singleton;
+
+public class ZKTimestampStorageModule extends AbstractModule {
+
+
+ @Override
+ public void configure() {
+
+ requireBinding(CuratorFramework.class);
+ // Timestamp storage creation
+ bind(TimestampStorage.class).to(ZKTimestampStorage.class).in(Singleton.class);
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/timestamp-storage/src/test/java/com/yahoo/omid/timestamp/storage/TestHBaseTimestampStorage.java
----------------------------------------------------------------------
diff --git a/timestamp-storage/src/test/java/com/yahoo/omid/timestamp/storage/TestHBaseTimestampStorage.java b/timestamp-storage/src/test/java/com/yahoo/omid/timestamp/storage/TestHBaseTimestampStorage.java
deleted file mode 100644
index c273995..0000000
--- a/timestamp-storage/src/test/java/com/yahoo/omid/timestamp/storage/TestHBaseTimestampStorage.java
+++ /dev/null
@@ -1,136 +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.timestamp.storage;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.hadoop.hbase.HBaseTestingUtility;
-import org.apache.hadoop.hbase.HColumnDescriptor;
-import org.apache.hadoop.hbase.HTableDescriptor;
-import org.apache.hadoop.hbase.MiniHBaseCluster;
-import org.apache.hadoop.hbase.TableName;
-import org.apache.hadoop.hbase.client.HBaseAdmin;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.testng.Assert;
-import org.testng.annotations.AfterClass;
-import org.testng.annotations.AfterMethod;
-import org.testng.annotations.BeforeClass;
-import org.testng.annotations.BeforeMethod;
-import org.testng.annotations.Test;
-
-import java.io.IOException;
-
-import static com.yahoo.omid.timestamp.storage.HBaseTimestampStorageConfig.*;
-import static org.testng.AssertJUnit.assertEquals;
-
-public class TestHBaseTimestampStorage {
-
- private static final Logger LOG = LoggerFactory.getLogger(TestHBaseTimestampStorage.class);
-
- private static final TableName TABLE_NAME = TableName.valueOf(DEFAULT_TIMESTAMP_STORAGE_TABLE_NAME);
-
- private static HBaseTestingUtility testutil;
- private static MiniHBaseCluster hbasecluster;
- protected static Configuration hbaseConf;
-
-
- @BeforeClass
- public static void setUpClass() throws Exception {
- // HBase setup
- hbaseConf = HBaseConfiguration.create();
-
- LOG.info("Create hbase");
- testutil = new HBaseTestingUtility(hbaseConf);
- hbasecluster = testutil.startMiniCluster(1);
-
- }
-
- @AfterClass
- public static void tearDownClass() throws Exception {
- if (hbasecluster != null) {
- testutil.shutdownMiniCluster();
- }
- }
-
- @BeforeMethod
- public void setUp() throws Exception {
- HBaseAdmin admin = testutil.getHBaseAdmin();
-
- if (!admin.tableExists(DEFAULT_TIMESTAMP_STORAGE_TABLE_NAME)) {
- HTableDescriptor desc = new HTableDescriptor(TABLE_NAME);
- HColumnDescriptor datafam = new HColumnDescriptor(DEFAULT_TIMESTAMP_STORAGE_CF_NAME);
- datafam.setMaxVersions(Integer.MAX_VALUE);
- desc.addFamily(datafam);
-
- admin.createTable(desc);
- }
-
- if (admin.isTableDisabled(DEFAULT_TIMESTAMP_STORAGE_TABLE_NAME)) {
- admin.enableTable(DEFAULT_TIMESTAMP_STORAGE_TABLE_NAME);
- }
- HTableDescriptor[] tables = admin.listTables();
- for (HTableDescriptor t : tables) {
- LOG.info(t.getNameAsString());
- }
- }
-
- @AfterMethod
- public void tearDown() {
- try {
- LOG.info("tearing Down");
- HBaseAdmin admin = testutil.getHBaseAdmin();
- admin.disableTable(DEFAULT_TIMESTAMP_STORAGE_TABLE_NAME);
- admin.deleteTable(DEFAULT_TIMESTAMP_STORAGE_TABLE_NAME);
-
- } catch (Exception e) {
- LOG.error("Error tearing down", e);
- }
- }
-
- @Test
- public void testHBaseTimestampStorage() throws Exception {
-
- final long INITIAL_TS_VALUE = 0;
- HBaseTimestampStorageConfig config = new HBaseTimestampStorageConfig();
- HBaseTimestampStorage tsStorage = new HBaseTimestampStorage(hbaseConf, config);
-
- // Test that the first time we get the timestamp is the initial value
- assertEquals("Initial value should be " + INITIAL_TS_VALUE, INITIAL_TS_VALUE, tsStorage.getMaxTimestamp());
-
- // Test that updating the timestamp succeeds when passing the initial value as the previous one
- long newTimestamp = 1;
- tsStorage.updateMaxTimestamp(INITIAL_TS_VALUE, newTimestamp);
-
- // Test setting a new timestamp fails (exception is thrown) when passing a wrong previous max timestamp
- long wrongTimestamp = 20;
- try {
- tsStorage.updateMaxTimestamp(wrongTimestamp, newTimestamp);
- Assert.fail("Shouldn't update");
- } catch (IOException e) {
- // Correct behavior
- }
- assertEquals("Value should be still " + newTimestamp, newTimestamp, tsStorage.getMaxTimestamp());
-
- // Test we can set a new timestamp when passing the right previous max timestamp
- long veryNewTimestamp = 40;
- tsStorage.updateMaxTimestamp(newTimestamp, veryNewTimestamp);
- assertEquals("Value should be " + veryNewTimestamp, veryNewTimestamp, tsStorage.getMaxTimestamp());
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/timestamp-storage/src/test/java/com/yahoo/omid/timestamp/storage/TestZKTimestampStorage.java
----------------------------------------------------------------------
diff --git a/timestamp-storage/src/test/java/com/yahoo/omid/timestamp/storage/TestZKTimestampStorage.java b/timestamp-storage/src/test/java/com/yahoo/omid/timestamp/storage/TestZKTimestampStorage.java
deleted file mode 100644
index 363768f..0000000
--- a/timestamp-storage/src/test/java/com/yahoo/omid/timestamp/storage/TestZKTimestampStorage.java
+++ /dev/null
@@ -1,241 +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.timestamp.storage;
-
-import org.apache.curator.RetryPolicy;
-import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.CuratorFrameworkFactory;
-import org.apache.curator.retry.ExponentialBackoffRetry;
-import org.apache.curator.test.TestingServer;
-import org.apache.curator.utils.CloseableUtils;
-import org.apache.zookeeper.ZooKeeper;
-import org.apache.zookeeper.data.Stat;
-import org.mockito.Mockito;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.testng.annotations.AfterMethod;
-import org.testng.annotations.BeforeMethod;
-import org.testng.annotations.Test;
-
-import java.io.IOException;
-import java.net.BindException;
-
-import static com.yahoo.omid.timestamp.storage.ZKTimestampPaths.TIMESTAMP_ZNODE;
-import static com.yahoo.omid.timestamp.storage.ZKTimestampStorage.INITIAL_MAX_TS_VALUE;
-import static org.mockito.Mockito.doThrow;
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.fail;
-
-public class TestZKTimestampStorage {
-
- private static final Logger LOG = LoggerFactory.getLogger(TestZKTimestampStorage.class);
-
- private static final int BYTES_IN_LONG = 8;
-
- private static final int ZK_PORT = 16666;
- private static final String ZK_CLUSTER = "localhost:" + ZK_PORT;
-
- private static final long NEGATIVE_TS = -1;
-
- private static final int ITERATION_COUNT = 10;
-
- private TestingServer zkServer;
-
- private CuratorFramework zkClient;
-
- private ZKTimestampStorage storage;
-
- private CuratorFramework storageInternalZKClient;
-
- @BeforeMethod
- public void initStuff() throws Exception {
- LOG.info("Creating ZK server instance listening in port {}...", ZK_PORT);
- while (zkServer == null) {
- try {
- zkServer = new TestingServer(ZK_PORT);
- } catch (BindException e) {
- System.err.println("Getting bind exception - retrying to allocate server");
- zkServer = null;
- }
- }
- LOG.info("ZK Server Started @ {}", zkServer.getConnectString());
-
- RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
-
- LOG.info("Creating Zookeeper Client connected to {}", ZK_CLUSTER);
- zkClient = CuratorFrameworkFactory.builder()
- .namespace("omid")
- .connectString(ZK_CLUSTER)
- .retryPolicy(retryPolicy)
- .connectionTimeoutMs(10) // Low timeout for tests
- .build();
- zkClient.start();
- zkClient.blockUntilConnected();
-
- LOG.info("Creating Internal Zookeeper Client connected to {}", ZK_CLUSTER);
- storageInternalZKClient = Mockito.spy(CuratorFrameworkFactory.builder()
- .namespace("omid")
- .connectString(ZK_CLUSTER)
- .retryPolicy(retryPolicy)
- .connectionTimeoutMs(10) // Low timeout for tests
- .build());
- storageInternalZKClient.start();
- storageInternalZKClient.blockUntilConnected();
-
- storage = new ZKTimestampStorage(storageInternalZKClient);
- }
-
- @AfterMethod
- public void closeStuff() throws Exception {
-
- CloseableUtils.closeQuietly(zkClient);
- LOG.info("ZK Client state {}", zkClient.getState());
- zkClient = null;
-
- CloseableUtils.closeQuietly(storageInternalZKClient);
- LOG.info("ZK Internal Client state {}", storageInternalZKClient.getState());
- storageInternalZKClient = null;
-
- CloseableUtils.closeQuietly(zkServer);
- LOG.info("ZK Server Stopped");
- zkServer = null;
-
- }
-
- @Test
- public void testBasicFunctionality() throws Exception {
-
- // Check ZNode for timestamp exists (storage instantiation should create it)
- Stat zNodeStats = zkClient.checkExists().forPath(TIMESTAMP_ZNODE);
- assertEquals(zNodeStats.getVersion(), 0);
-
- // Initial checks
- assertEquals(storage.getMaxTimestamp(), INITIAL_MAX_TS_VALUE);
- byte[] data = zkClient.getData().forPath(TIMESTAMP_ZNODE);
- assertEquals(data.length, BYTES_IN_LONG);
-
- // Check new timestamp does not allow negative values...
- try {
- storage.updateMaxTimestamp(INITIAL_MAX_TS_VALUE, NEGATIVE_TS);
- fail();
- } catch (IllegalArgumentException e) {
- // Expected exception
- }
-
- // ...nor is less than previous timestamp
- try {
- storage.updateMaxTimestamp(1, 0);
- fail();
- } catch (IllegalArgumentException e) {
- // Expected exception
- }
-
- // Check that the original version is still there
- zNodeStats = zkClient.checkExists().forPath(TIMESTAMP_ZNODE);
- assertEquals(zNodeStats.getVersion(), 0);
-
- // Iterate updating the timestamp and check the final value
- long previousMaxTimestamp = INITIAL_MAX_TS_VALUE;
- for (int i = 0; i < ITERATION_COUNT; i++) {
- long newMaxTimestamp = previousMaxTimestamp + 1_000_000;
- storage.updateMaxTimestamp(previousMaxTimestamp, newMaxTimestamp);
- previousMaxTimestamp = newMaxTimestamp;
- }
- assertEquals(storage.getMaxTimestamp(), 1_000_000 * ITERATION_COUNT);
- // Check the znode version has changed accordingly
- zNodeStats = zkClient.checkExists().forPath(TIMESTAMP_ZNODE);
- assertEquals(zNodeStats.getVersion(), ITERATION_COUNT);
-
- // Check exceptions
- doThrow(new RuntimeException()).when(storageInternalZKClient).getData();
- try {
- storage.getMaxTimestamp();
- fail();
- } catch (IOException e) {
- // Expected exception
- }
-
- doThrow(new RuntimeException()).when(storageInternalZKClient).setData();
- try {
- storage.updateMaxTimestamp(INITIAL_MAX_TS_VALUE, INITIAL_MAX_TS_VALUE + 1_000_000);
- fail();
- } catch (IOException e) {
- // Expected exception
- }
-
- // Reset the mock and double-check last result
- Mockito.reset(storageInternalZKClient);
- assertEquals(storage.getMaxTimestamp(), 1_000_000 * ITERATION_COUNT);
-
- // Finally check the znode version is still the same
- zNodeStats = zkClient.checkExists().forPath(TIMESTAMP_ZNODE);
- assertEquals(zNodeStats.getVersion(), ITERATION_COUNT);
- }
-
- @Test
- public void testZkClientWhenZKIsDownAndRestarts() throws Exception {
-
- // Iterate updating the timestamp and check the final value
- long previousMaxTimestamp = INITIAL_MAX_TS_VALUE;
- for (int i = 0; i < ITERATION_COUNT; i++) {
- long newMaxTimestamp = previousMaxTimestamp + 1_000_000;
- storage.updateMaxTimestamp(previousMaxTimestamp, newMaxTimestamp);
- previousMaxTimestamp = newMaxTimestamp;
- }
- assertEquals(storage.getMaxTimestamp(), 1_000_000 * ITERATION_COUNT);
-
- // Stop ZK Server, expect the IO exception, reconnect and get the right value
- LOG.info("Stopping ZK Server");
- zkServer.stop();
- LOG.info("ZK Server Stopped");
-
- try {
- storage.getMaxTimestamp();
- fail();
- } catch (IOException ioe) {
- // Expected exception
- }
-
- LOG.info("Restarting ZK again");
- zkServer.restart();
- assertEquals(storage.getMaxTimestamp(), 1_000_000 * ITERATION_COUNT);
-
- }
-
- @Test
- public void testZkClientLosingSession() throws Exception {
-
- // Cut the session in the server through the client
- long sessionId = zkClient.getZookeeperClient().getZooKeeper().getSessionId();
- byte[] sessionPasswd = zkClient.getZookeeperClient().getZooKeeper().getSessionPasswd();
- ZooKeeper zk = new ZooKeeper(ZK_CLUSTER, 1000, null, sessionId, sessionPasswd);
- zk.close();
- LOG.info("ZKClient session closed");
-
- // Iterate updating the timestamp and check the final value
- long previousMaxTimestamp = INITIAL_MAX_TS_VALUE;
- for (int i = 0; i < ITERATION_COUNT; i++) {
- long newMaxTimestamp = previousMaxTimestamp + 1_000_000;
- storage.updateMaxTimestamp(previousMaxTimestamp, newMaxTimestamp);
- LOG.info("Updating timestamp. Previous/New {}/{}", previousMaxTimestamp, newMaxTimestamp);
- previousMaxTimestamp = newMaxTimestamp;
- }
- assertEquals(storage.getMaxTimestamp(), 1_000_000 * ITERATION_COUNT);
-
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/timestamp-storage/src/test/java/org/apache/omid/timestamp/storage/TestHBaseTimestampStorage.java
----------------------------------------------------------------------
diff --git a/timestamp-storage/src/test/java/org/apache/omid/timestamp/storage/TestHBaseTimestampStorage.java b/timestamp-storage/src/test/java/org/apache/omid/timestamp/storage/TestHBaseTimestampStorage.java
new file mode 100644
index 0000000..4218734
--- /dev/null
+++ b/timestamp-storage/src/test/java/org/apache/omid/timestamp/storage/TestHBaseTimestampStorage.java
@@ -0,0 +1,136 @@
+/**
+ * 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.timestamp.storage;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.MiniHBaseCluster;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.HBaseAdmin;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.io.IOException;
+
+import static org.apache.omid.timestamp.storage.HBaseTimestampStorageConfig.*;
+import static org.testng.AssertJUnit.assertEquals;
+
+public class TestHBaseTimestampStorage {
+
+ private static final Logger LOG = LoggerFactory.getLogger(TestHBaseTimestampStorage.class);
+
+ private static final TableName TABLE_NAME = TableName.valueOf(DEFAULT_TIMESTAMP_STORAGE_TABLE_NAME);
+
+ private static HBaseTestingUtility testutil;
+ private static MiniHBaseCluster hbasecluster;
+ protected static Configuration hbaseConf;
+
+
+ @BeforeClass
+ public static void setUpClass() throws Exception {
+ // HBase setup
+ hbaseConf = HBaseConfiguration.create();
+
+ LOG.info("Create hbase");
+ testutil = new HBaseTestingUtility(hbaseConf);
+ hbasecluster = testutil.startMiniCluster(1);
+
+ }
+
+ @AfterClass
+ public static void tearDownClass() throws Exception {
+ if (hbasecluster != null) {
+ testutil.shutdownMiniCluster();
+ }
+ }
+
+ @BeforeMethod
+ public void setUp() throws Exception {
+ HBaseAdmin admin = testutil.getHBaseAdmin();
+
+ if (!admin.tableExists(DEFAULT_TIMESTAMP_STORAGE_TABLE_NAME)) {
+ HTableDescriptor desc = new HTableDescriptor(TABLE_NAME);
+ HColumnDescriptor datafam = new HColumnDescriptor(DEFAULT_TIMESTAMP_STORAGE_CF_NAME);
+ datafam.setMaxVersions(Integer.MAX_VALUE);
+ desc.addFamily(datafam);
+
+ admin.createTable(desc);
+ }
+
+ if (admin.isTableDisabled(DEFAULT_TIMESTAMP_STORAGE_TABLE_NAME)) {
+ admin.enableTable(DEFAULT_TIMESTAMP_STORAGE_TABLE_NAME);
+ }
+ HTableDescriptor[] tables = admin.listTables();
+ for (HTableDescriptor t : tables) {
+ LOG.info(t.getNameAsString());
+ }
+ }
+
+ @AfterMethod
+ public void tearDown() {
+ try {
+ LOG.info("tearing Down");
+ HBaseAdmin admin = testutil.getHBaseAdmin();
+ admin.disableTable(DEFAULT_TIMESTAMP_STORAGE_TABLE_NAME);
+ admin.deleteTable(DEFAULT_TIMESTAMP_STORAGE_TABLE_NAME);
+
+ } catch (Exception e) {
+ LOG.error("Error tearing down", e);
+ }
+ }
+
+ @Test
+ public void testHBaseTimestampStorage() throws Exception {
+
+ final long INITIAL_TS_VALUE = 0;
+ HBaseTimestampStorageConfig config = new HBaseTimestampStorageConfig();
+ HBaseTimestampStorage tsStorage = new HBaseTimestampStorage(hbaseConf, config);
+
+ // Test that the first time we get the timestamp is the initial value
+ assertEquals("Initial value should be " + INITIAL_TS_VALUE, INITIAL_TS_VALUE, tsStorage.getMaxTimestamp());
+
+ // Test that updating the timestamp succeeds when passing the initial value as the previous one
+ long newTimestamp = 1;
+ tsStorage.updateMaxTimestamp(INITIAL_TS_VALUE, newTimestamp);
+
+ // Test setting a new timestamp fails (exception is thrown) when passing a wrong previous max timestamp
+ long wrongTimestamp = 20;
+ try {
+ tsStorage.updateMaxTimestamp(wrongTimestamp, newTimestamp);
+ Assert.fail("Shouldn't update");
+ } catch (IOException e) {
+ // Correct behavior
+ }
+ assertEquals("Value should be still " + newTimestamp, newTimestamp, tsStorage.getMaxTimestamp());
+
+ // Test we can set a new timestamp when passing the right previous max timestamp
+ long veryNewTimestamp = 40;
+ tsStorage.updateMaxTimestamp(newTimestamp, veryNewTimestamp);
+ assertEquals("Value should be " + veryNewTimestamp, veryNewTimestamp, tsStorage.getMaxTimestamp());
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/timestamp-storage/src/test/java/org/apache/omid/timestamp/storage/TestZKTimestampStorage.java
----------------------------------------------------------------------
diff --git a/timestamp-storage/src/test/java/org/apache/omid/timestamp/storage/TestZKTimestampStorage.java b/timestamp-storage/src/test/java/org/apache/omid/timestamp/storage/TestZKTimestampStorage.java
new file mode 100644
index 0000000..1e370e2
--- /dev/null
+++ b/timestamp-storage/src/test/java/org/apache/omid/timestamp/storage/TestZKTimestampStorage.java
@@ -0,0 +1,241 @@
+/**
+ * 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.timestamp.storage;
+
+import org.apache.curator.RetryPolicy;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.ExponentialBackoffRetry;
+import org.apache.curator.test.TestingServer;
+import org.apache.curator.utils.CloseableUtils;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.Stat;
+import org.mockito.Mockito;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.io.IOException;
+import java.net.BindException;
+
+import static org.apache.omid.timestamp.storage.ZKTimestampPaths.TIMESTAMP_ZNODE;
+import static org.apache.omid.timestamp.storage.ZKTimestampStorage.INITIAL_MAX_TS_VALUE;
+import static org.mockito.Mockito.doThrow;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.fail;
+
+public class TestZKTimestampStorage {
+
+ private static final Logger LOG = LoggerFactory.getLogger(TestZKTimestampStorage.class);
+
+ private static final int BYTES_IN_LONG = 8;
+
+ private static final int ZK_PORT = 16666;
+ private static final String ZK_CLUSTER = "localhost:" + ZK_PORT;
+
+ private static final long NEGATIVE_TS = -1;
+
+ private static final int ITERATION_COUNT = 10;
+
+ private TestingServer zkServer;
+
+ private CuratorFramework zkClient;
+
+ private ZKTimestampStorage storage;
+
+ private CuratorFramework storageInternalZKClient;
+
+ @BeforeMethod
+ public void initStuff() throws Exception {
+ LOG.info("Creating ZK server instance listening in port {}...", ZK_PORT);
+ while (zkServer == null) {
+ try {
+ zkServer = new TestingServer(ZK_PORT);
+ } catch (BindException e) {
+ System.err.println("Getting bind exception - retrying to allocate server");
+ zkServer = null;
+ }
+ }
+ LOG.info("ZK Server Started @ {}", zkServer.getConnectString());
+
+ RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
+
+ LOG.info("Creating Zookeeper Client connected to {}", ZK_CLUSTER);
+ zkClient = CuratorFrameworkFactory.builder()
+ .namespace("omid")
+ .connectString(ZK_CLUSTER)
+ .retryPolicy(retryPolicy)
+ .connectionTimeoutMs(10) // Low timeout for tests
+ .build();
+ zkClient.start();
+ zkClient.blockUntilConnected();
+
+ LOG.info("Creating Internal Zookeeper Client connected to {}", ZK_CLUSTER);
+ storageInternalZKClient = Mockito.spy(CuratorFrameworkFactory.builder()
+ .namespace("omid")
+ .connectString(ZK_CLUSTER)
+ .retryPolicy(retryPolicy)
+ .connectionTimeoutMs(10) // Low timeout for tests
+ .build());
+ storageInternalZKClient.start();
+ storageInternalZKClient.blockUntilConnected();
+
+ storage = new ZKTimestampStorage(storageInternalZKClient);
+ }
+
+ @AfterMethod
+ public void closeStuff() throws Exception {
+
+ CloseableUtils.closeQuietly(zkClient);
+ LOG.info("ZK Client state {}", zkClient.getState());
+ zkClient = null;
+
+ CloseableUtils.closeQuietly(storageInternalZKClient);
+ LOG.info("ZK Internal Client state {}", storageInternalZKClient.getState());
+ storageInternalZKClient = null;
+
+ CloseableUtils.closeQuietly(zkServer);
+ LOG.info("ZK Server Stopped");
+ zkServer = null;
+
+ }
+
+ @Test
+ public void testBasicFunctionality() throws Exception {
+
+ // Check ZNode for timestamp exists (storage instantiation should create it)
+ Stat zNodeStats = zkClient.checkExists().forPath(TIMESTAMP_ZNODE);
+ assertEquals(zNodeStats.getVersion(), 0);
+
+ // Initial checks
+ assertEquals(storage.getMaxTimestamp(), INITIAL_MAX_TS_VALUE);
+ byte[] data = zkClient.getData().forPath(TIMESTAMP_ZNODE);
+ assertEquals(data.length, BYTES_IN_LONG);
+
+ // Check new timestamp does not allow negative values...
+ try {
+ storage.updateMaxTimestamp(INITIAL_MAX_TS_VALUE, NEGATIVE_TS);
+ fail();
+ } catch (IllegalArgumentException e) {
+ // Expected exception
+ }
+
+ // ...nor is less than previous timestamp
+ try {
+ storage.updateMaxTimestamp(1, 0);
+ fail();
+ } catch (IllegalArgumentException e) {
+ // Expected exception
+ }
+
+ // Check that the original version is still there
+ zNodeStats = zkClient.checkExists().forPath(TIMESTAMP_ZNODE);
+ assertEquals(zNodeStats.getVersion(), 0);
+
+ // Iterate updating the timestamp and check the final value
+ long previousMaxTimestamp = INITIAL_MAX_TS_VALUE;
+ for (int i = 0; i < ITERATION_COUNT; i++) {
+ long newMaxTimestamp = previousMaxTimestamp + 1_000_000;
+ storage.updateMaxTimestamp(previousMaxTimestamp, newMaxTimestamp);
+ previousMaxTimestamp = newMaxTimestamp;
+ }
+ assertEquals(storage.getMaxTimestamp(), 1_000_000 * ITERATION_COUNT);
+ // Check the znode version has changed accordingly
+ zNodeStats = zkClient.checkExists().forPath(TIMESTAMP_ZNODE);
+ assertEquals(zNodeStats.getVersion(), ITERATION_COUNT);
+
+ // Check exceptions
+ doThrow(new RuntimeException()).when(storageInternalZKClient).getData();
+ try {
+ storage.getMaxTimestamp();
+ fail();
+ } catch (IOException e) {
+ // Expected exception
+ }
+
+ doThrow(new RuntimeException()).when(storageInternalZKClient).setData();
+ try {
+ storage.updateMaxTimestamp(INITIAL_MAX_TS_VALUE, INITIAL_MAX_TS_VALUE + 1_000_000);
+ fail();
+ } catch (IOException e) {
+ // Expected exception
+ }
+
+ // Reset the mock and double-check last result
+ Mockito.reset(storageInternalZKClient);
+ assertEquals(storage.getMaxTimestamp(), 1_000_000 * ITERATION_COUNT);
+
+ // Finally check the znode version is still the same
+ zNodeStats = zkClient.checkExists().forPath(TIMESTAMP_ZNODE);
+ assertEquals(zNodeStats.getVersion(), ITERATION_COUNT);
+ }
+
+ @Test
+ public void testZkClientWhenZKIsDownAndRestarts() throws Exception {
+
+ // Iterate updating the timestamp and check the final value
+ long previousMaxTimestamp = INITIAL_MAX_TS_VALUE;
+ for (int i = 0; i < ITERATION_COUNT; i++) {
+ long newMaxTimestamp = previousMaxTimestamp + 1_000_000;
+ storage.updateMaxTimestamp(previousMaxTimestamp, newMaxTimestamp);
+ previousMaxTimestamp = newMaxTimestamp;
+ }
+ assertEquals(storage.getMaxTimestamp(), 1_000_000 * ITERATION_COUNT);
+
+ // Stop ZK Server, expect the IO exception, reconnect and get the right value
+ LOG.info("Stopping ZK Server");
+ zkServer.stop();
+ LOG.info("ZK Server Stopped");
+
+ try {
+ storage.getMaxTimestamp();
+ fail();
+ } catch (IOException ioe) {
+ // Expected exception
+ }
+
+ LOG.info("Restarting ZK again");
+ zkServer.restart();
+ assertEquals(storage.getMaxTimestamp(), 1_000_000 * ITERATION_COUNT);
+
+ }
+
+ @Test
+ public void testZkClientLosingSession() throws Exception {
+
+ // Cut the session in the server through the client
+ long sessionId = zkClient.getZookeeperClient().getZooKeeper().getSessionId();
+ byte[] sessionPasswd = zkClient.getZookeeperClient().getZooKeeper().getSessionPasswd();
+ ZooKeeper zk = new ZooKeeper(ZK_CLUSTER, 1000, null, sessionId, sessionPasswd);
+ zk.close();
+ LOG.info("ZKClient session closed");
+
+ // Iterate updating the timestamp and check the final value
+ long previousMaxTimestamp = INITIAL_MAX_TS_VALUE;
+ for (int i = 0; i < ITERATION_COUNT; i++) {
+ long newMaxTimestamp = previousMaxTimestamp + 1_000_000;
+ storage.updateMaxTimestamp(previousMaxTimestamp, newMaxTimestamp);
+ LOG.info("Updating timestamp. Previous/New {}/{}", previousMaxTimestamp, newMaxTimestamp);
+ previousMaxTimestamp = newMaxTimestamp;
+ }
+ assertEquals(storage.getMaxTimestamp(), 1_000_000 * ITERATION_COUNT);
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/timestamp-storage/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/timestamp-storage/src/test/resources/log4j.properties b/timestamp-storage/src/test/resources/log4j.properties
index 23b950e..3f565a2 100644
--- a/timestamp-storage/src/test/resources/log4j.properties
+++ b/timestamp-storage/src/test/resources/log4j.properties
@@ -39,14 +39,13 @@ log4j.logger.org.apache.zookeeper=ERROR
log4j.logger.org.apache.bookkeeper=FATAL
log4j.logger.org.apache.hadoop.hbase=ERROR
log4j.logger.org.apache.hadoop.ipc=ERROR
-
-log4j.logger.com.yahoo.omid.util=DEBUG
-#log4j.logger.com.yahoo.omid.regionserver.TransactionalRegionServer=TRACE
-#log4j.logger.com.yahoo.omid.TestBasicTransaction=TRACE
-#log4j.logger.com.yahoo.omid.client.TSOClient=TRACE
-#log4j.logger.com.yahoo.omid.client.TransactionState=TRACE
-#log4j.logger.com.yahoo.omid.OmidTestBase=TRACE
-#log4j.logger.com.yahoo.omid.tso.ThroughputMonitor=INFO
+log4j.logger.org.apache.omid.util=DEBUG
+#log4j.logger.org.apache.omid.regionserver.TransactionalRegionServer=TRACE
+#log4j.logger.org.apache.omid.TestBasicTransaction=TRACE
+#log4j.logger.org.apache.omid.client.TSOClient=TRACE
+#log4j.logger.org.apache.omid.client.TransactionState=TRACE
+#log4j.logger.org.apache.omid.OmidTestBase=TRACE
+#log4j.logger.org.apache.omid.tso.ThroughputMonitor=INFO
#log4j.logger.org.apache.hadoop.fs.FSNamesystem=DEBUG
# Make these two classes INFO-level. Make them DEBUG to see more zk debug.
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/transaction-client/pom.xml
----------------------------------------------------------------------
diff --git a/transaction-client/pom.xml b/transaction-client/pom.xml
index f7c27d9..e0102a5 100644
--- a/transaction-client/pom.xml
+++ b/transaction-client/pom.xml
@@ -3,7 +3,7 @@
<modelVersion>4.0.0</modelVersion>
<parent>
- <groupId>com.yahoo.omid</groupId>
+ <groupId>org.apache.omid</groupId>
<artifactId>omid</artifactId>
<version>0.8.1.37-SNAPSHOT</version>
</parent>
@@ -17,27 +17,27 @@
<!-- Dependencies on Omid modules -->
<dependency>
- <groupId>com.yahoo.omid</groupId>
+ <groupId>org.apache.omid</groupId>
<artifactId>common</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
- <groupId>com.yahoo.omid</groupId>
+ <groupId>org.apache.omid</groupId>
<artifactId>statemachine</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
- <groupId>com.yahoo.omid</groupId>
+ <groupId>org.apache.omid</groupId>
<artifactId>commit-table</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
- <groupId>com.yahoo.omid</groupId>
+ <groupId>org.apache.omid</groupId>
<artifactId>metrics</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
- <groupId>com.yahoo.omid</groupId>
+ <groupId>org.apache.omid</groupId>
<artifactId>commit-table</artifactId>
<version>${project.version}</version>
<scope>test</scope>
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/transaction-client/src/main/java/com/yahoo/omid/transaction/AbstractTransaction.java
----------------------------------------------------------------------
diff --git a/transaction-client/src/main/java/com/yahoo/omid/transaction/AbstractTransaction.java b/transaction-client/src/main/java/com/yahoo/omid/transaction/AbstractTransaction.java
deleted file mode 100644
index f09772a..0000000
--- a/transaction-client/src/main/java/com/yahoo/omid/transaction/AbstractTransaction.java
+++ /dev/null
@@ -1,217 +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.Optional;
-import com.yahoo.omid.tso.client.CellId;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-/**
- * Omid's base abstract implementation of the {@link Transaction} interface.
- * Provides extra methods to access other basic transaction state required by
- * {@link TransactionManagerExtension} implementations based on snapshot
- * isolation.
- *
- * So, this abstract class must be extended by particular implementations of
- * transaction managers related to different storage systems (HBase...)
- */
-public abstract class AbstractTransaction<T extends CellId> implements Transaction {
-
- private transient Map<String, Object> metadata = new HashMap<>();
- private final AbstractTransactionManager transactionManager;
- private final long startTimestamp;
- private final long epoch;
- private long commitTimestamp;
- private boolean isRollbackOnly;
- private final Set<T> writeSet;
- private Status status = Status.RUNNING;
-
- /**
- * Base constructor
- *
- * @param transactionId
- * transaction identifier to assign
- * @param epoch
- * epoch of the TSOServer instance that created this transaction
- * Used in High Availability to guarantee data consistency
- * @param writeSet
- * initial write set for the transaction.
- * Should be empty in most cases.
- * @param transactionManager
- * transaction manager associated to this transaction.
- * Usually, should be the one that created the transaction
- * instance.
- */
- public AbstractTransaction(long transactionId,
- long epoch,
- Set<T> writeSet,
- AbstractTransactionManager transactionManager) {
- this.startTimestamp = transactionId;
- this.epoch = epoch;
- this.writeSet = writeSet;
- this.transactionManager = transactionManager;
- }
-
- /**
- * Allows to define specific clean-up task for transaction implementations
- */
- public abstract void cleanup();
-
- /**
- * @see com.yahoo.omid.transaction.Transaction#getTransactionId()
- */
- @Override
- public long getTransactionId() {
- return startTimestamp;
- }
-
- /**
- * @see com.yahoo.omid.transaction.Transaction#getEpoch()
- */
- @Override
- public long getEpoch() {
- return epoch;
- }
-
- /**
- * @see com.yahoo.omid.transaction.Transaction#getStatus()
- */
- @Override
- public Status getStatus() {
- return status;
- }
-
- /**
- * @see Transaction#isRollbackOnly()
- */
- @Override
- public void setRollbackOnly() {
- isRollbackOnly = true;
- }
-
- /**
- * @see com.yahoo.omid.transaction.Transaction#isRollbackOnly()
- */
- @Override
- public boolean isRollbackOnly() {
- return isRollbackOnly;
- }
-
- /**
- * Returns transaction manager associated to this transaction.
- * @return transaction manager
- */
- public AbstractTransactionManager getTransactionManager() {
- return transactionManager;
- }
-
- /**
- * Returns the start timestamp for this transaction.
- * @return start timestamp
- */
- public long getStartTimestamp() {
- return startTimestamp;
- }
-
- /**
- * Returns the commit timestamp for this transaction.
- * @return commit timestamp
- */
- public long getCommitTimestamp() {
- return commitTimestamp;
- }
-
- /**
- * Sets the commit timestamp for this transaction.
- * @param commitTimestamp
- * the commit timestamp to set
- */
- public void setCommitTimestamp(long commitTimestamp) {
- this.commitTimestamp = commitTimestamp;
- }
-
- /**
- * Sets the status for this transaction.
- * @param status
- * the {@link Status} to set
- */
- public void setStatus(Status status) {
- this.status = status;
- }
-
- /**
- * Returns the current write-set for this transaction.
- * @return write set
- */
- public Set<T> getWriteSet() {
- return writeSet;
- }
-
- /**
- * Adds an element to the transaction write-set.
- * @param element
- * the element to add
- */
- public void addWriteSetElement(T element) {
- writeSet.add(element);
- }
-
- @Override
- public String toString() {
- return String.format("Tx-%s [%s] (ST=%d, CT=%d, Epoch=%d) WriteSet %s",
- Long.toHexString(getTransactionId()),
- status,
- startTimestamp,
- commitTimestamp,
- epoch,
- writeSet);
- }
-
- @Override
- public Optional<Object> getMetadata(String key) {
- return Optional.fromNullable(metadata.get(key));
- }
-
- /**
- * Expects they metadata stored under key "key" to be of the "Set" type,
- * append "value" to the existing set or creates a new one
- */
- @Override
- @SuppressWarnings("unchecked")
- public void appendMetadata(String key, Object value) {
- List existingValue = (List) metadata.get(key);
- if (existingValue == null) {
- List<Object> newList = new ArrayList<>();
- newList.add(value);
- metadata.put(key, newList);
- } else {
- existingValue.add(value);
- }
- }
-
- @Override
- public void setMetadata(String key, Object value) {
- metadata.put(key, value);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/transaction-client/src/main/java/com/yahoo/omid/transaction/AbstractTransactionManager.java
----------------------------------------------------------------------
diff --git a/transaction-client/src/main/java/com/yahoo/omid/transaction/AbstractTransactionManager.java b/transaction-client/src/main/java/com/yahoo/omid/transaction/AbstractTransactionManager.java
deleted file mode 100644
index 42bec31..0000000
--- a/transaction-client/src/main/java/com/yahoo/omid/transaction/AbstractTransactionManager.java
+++ /dev/null
@@ -1,502 +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.util.concurrent.Futures;
-import com.yahoo.omid.committable.CommitTable;
-import com.yahoo.omid.committable.CommitTable.CommitTimestamp;
-import com.yahoo.omid.metrics.Counter;
-import com.yahoo.omid.metrics.MetricsRegistry;
-import com.yahoo.omid.metrics.Timer;
-import com.yahoo.omid.transaction.Transaction.Status;
-import com.yahoo.omid.tso.client.AbortException;
-import com.yahoo.omid.tso.client.CellId;
-import com.yahoo.omid.tso.client.ConnectionException;
-import com.yahoo.omid.tso.client.NewTSOException;
-import com.yahoo.omid.tso.client.ServiceUnavailableException;
-import com.yahoo.omid.tso.client.TSOClient;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.util.concurrent.ExecutionException;
-
-import static com.yahoo.omid.committable.CommitTable.CommitTimestamp.Location.CACHE;
-import static com.yahoo.omid.committable.CommitTable.CommitTimestamp.Location.COMMIT_TABLE;
-import static com.yahoo.omid.committable.CommitTable.CommitTimestamp.Location.NOT_PRESENT;
-import static com.yahoo.omid.committable.CommitTable.CommitTimestamp.Location.SHADOW_CELL;
-import static com.yahoo.omid.metrics.MetricsUtils.name;
-
-/**
- * Omid's base abstract implementation of the {@link TransactionManager} interface.
- *
- * Provides extra methods to allow transaction manager developers to perform
- * different actions before/after the methods exposed by the {@link TransactionManager} interface.
- *
- * So, this abstract class must be extended by particular implementations of
- * transaction managers related to different storage systems (HBase...)
- */
-public abstract class AbstractTransactionManager implements TransactionManager {
-
- private static final Logger LOG = LoggerFactory.getLogger(AbstractTransactionManager.class);
-
- public interface TransactionFactory<T extends CellId> {
-
- AbstractTransaction<T> createTransaction(long transactionId, long epoch, AbstractTransactionManager tm);
-
- }
-
- private final PostCommitActions postCommitter;
- protected final TSOClient tsoClient;
- protected final CommitTable.Client commitTableClient;
- private final TransactionFactory<? extends CellId> transactionFactory;
-
- // Metrics
- private final Timer startTimestampTimer;
- private final Timer commitTimer;
- private final Counter committedTxsCounter;
- private final Counter rolledbackTxsCounter;
- private final Counter errorTxsCounter;
- private final Counter invalidatedTxsCounter;
-
- /**
- * Base constructor
- *
- * @param metrics
- * instrumentation metrics
- * @param postCommitter
- * post commit action executor
- * @param tsoClient
- * a client for accessing functionality of the status oracle
- * @param commitTableClient
- * a client for accessing functionality of the commit table
- * @param transactionFactory
- * a transaction factory to create the specific transaction
- * objects required by the transaction manager being implemented.
- */
- public AbstractTransactionManager(MetricsRegistry metrics,
- PostCommitActions postCommitter,
- TSOClient tsoClient,
- CommitTable.Client commitTableClient,
- TransactionFactory<? extends CellId> transactionFactory) {
-
- this.tsoClient = tsoClient;
- this.postCommitter = postCommitter;
- this.commitTableClient = commitTableClient;
- this.transactionFactory = transactionFactory;
-
- // Metrics configuration
- this.startTimestampTimer = metrics.timer(name("omid", "tm", "hbase", "startTimestamp", "latency"));
- this.commitTimer = metrics.timer(name("omid", "tm", "hbase", "commit", "latency"));
- this.committedTxsCounter = metrics.counter(name("omid", "tm", "hbase", "committedTxs"));
- this.rolledbackTxsCounter = metrics.counter(name("omid", "tm", "hbase", "rolledbackTxs"));
- this.errorTxsCounter = metrics.counter(name("omid", "tm", "hbase", "erroredTxs"));
- this.invalidatedTxsCounter = metrics.counter(name("omid", "tm", "hbase", "invalidatedTxs"));
-
- }
-
- /**
- * Allows transaction manager developers to perform actions before creating a transaction.
- * @throws TransactionManagerException
- */
- public void preBegin() throws TransactionManagerException {}
-
- /**
- * @see com.yahoo.omid.transaction.TransactionManager#begin()
- */
- @Override
- public final Transaction begin() throws TransactionException {
-
- try {
- preBegin();
-
- long startTimestamp, epoch;
-
- // The loop is required for HA scenarios where we get the timestamp
- // but when getting the epoch, the client is connected to a new TSOServer
- // When this happen, the epoch will be larger than the startTimestamp,
- // so we need to start the transaction again. We use the fact that epoch
- // is always smaller or equal to a timestamp, and therefore, we first need
- // to get the timestamp and then the epoch.
- startTimestampTimer.start();
- try {
- do {
- startTimestamp = tsoClient.getNewStartTimestamp().get();
- epoch = tsoClient.getEpoch();
- } while (epoch > startTimestamp);
- } finally {
- startTimestampTimer.stop();
- }
-
- AbstractTransaction<? extends CellId> tx = transactionFactory.createTransaction(startTimestamp, epoch, this);
-
- postBegin(tx);
-
- return tx;
- } catch (TransactionManagerException e) {
- throw new TransactionException("An error has occured during PreBegin/PostBegin", e);
- } catch (ExecutionException e) {
- throw new TransactionException("Could not get new timestamp", e);
- } catch (InterruptedException ie) {
- Thread.currentThread().interrupt();
- throw new TransactionException("Interrupted getting timestamp", ie);
- }
- }
-
- /**
- * Allows transaction manager developers to perform actions after having started a transaction.
- * @param transaction
- * the transaction that was just created.
- * @throws TransactionManagerException
- */
- public void postBegin(AbstractTransaction<? extends CellId> transaction) throws TransactionManagerException {}
-
- /**
- * Allows transaction manager developers to perform actions before committing a transaction.
- * @param transaction
- * the transaction that is going to be committed.
- * @throws TransactionManagerException
- */
- public void preCommit(AbstractTransaction<? extends CellId> transaction) throws TransactionManagerException {}
-
- /**
- * @see com.yahoo.omid.transaction.TransactionManager#commit(Transaction)
- */
- @Override
- public final void commit(Transaction transaction) throws RollbackException, TransactionException {
-
- AbstractTransaction<? extends CellId> tx = enforceAbstractTransactionAsParam(transaction);
- enforceTransactionIsInRunningState(tx);
-
- if (tx.isRollbackOnly()) { // Manage explicit user rollback
- rollback(tx);
- throw new RollbackException(tx + ": Tx was set to rollback explicitly");
- }
-
- try {
-
- preCommit(tx);
-
- commitTimer.start();
- try {
- if (tx.getWriteSet().isEmpty()) {
- markReadOnlyTransaction(tx); // No need for read-only transactions to contact the TSO Server
- } else {
- commitRegularTransaction(tx);
- }
- committedTxsCounter.inc();
- } finally {
- commitTimer.stop();
- }
-
- postCommit(tx);
-
- } catch (TransactionManagerException e) {
- throw new TransactionException(e.getMessage(), e);
- }
-
- }
-
- /**
- * Allows transaction manager developers to perform actions after committing a transaction.
- * @param transaction
- * the transaction that was committed.
- * @throws TransactionManagerException
- */
- public void postCommit(AbstractTransaction<? extends CellId> transaction) throws TransactionManagerException {}
-
- /**
- * Allows transaction manager developers to perform actions before rolling-back a transaction.
- * @param transaction
- * the transaction that is going to be rolled-back.
- * @throws TransactionManagerException
- */
- public void preRollback(AbstractTransaction<? extends CellId> transaction) throws TransactionManagerException {}
-
- /**
- * @see com.yahoo.omid.transaction.TransactionManager#rollback(Transaction)
- */
- @Override
- public final void rollback(Transaction transaction) throws TransactionException {
-
- AbstractTransaction<? extends CellId> tx = enforceAbstractTransactionAsParam(transaction);
- enforceTransactionIsInRunningState(tx);
-
- try {
-
- preRollback(tx);
-
- // Make sure its commit timestamp is 0, so the cleanup does the right job
- tx.setCommitTimestamp(0);
- tx.setStatus(Status.ROLLEDBACK);
-
- postRollback(tx);
-
- } catch (TransactionManagerException e) {
- throw new TransactionException(e.getMessage(), e);
- } finally {
- tx.cleanup();
- }
-
- }
-
- /**
- * Allows transaction manager developers to perform actions after rolling-back a transaction.
- * @param transaction
- * the transaction that was rolled-back.
- * @throws TransactionManagerException
- */
- public void postRollback(AbstractTransaction<? extends CellId> transaction) throws TransactionManagerException {}
-
- /**
- * Check if the transaction commit data is in the shadow cell
- * @param cellStartTimestamp
- * the transaction start timestamp
- * locator
- * the timestamp locator
- * @throws IOException
- */
- Optional<CommitTimestamp> readCommitTimestampFromShadowCell(long cellStartTimestamp, CommitTimestampLocator locator)
- throws IOException
- {
-
- Optional<CommitTimestamp> commitTS = Optional.absent();
-
- Optional<Long> commitTimestamp = locator.readCommitTimestampFromShadowCell(cellStartTimestamp);
- if (commitTimestamp.isPresent()) {
- commitTS = Optional.of(new CommitTimestamp(SHADOW_CELL, commitTimestamp.get(), true)); // Valid commit TS
- }
-
- return commitTS;
- }
-
- /**
- * This function returns the commit timestamp for a particular cell if the transaction was already committed in
- * the system. In case the transaction was not committed and the cell was written by transaction initialized by a
- * previous TSO server, an invalidation try occurs.
- * Otherwise the function returns a value that indicates that the commit timestamp was not found.
- * @param cellStartTimestamp
- * start timestamp of the cell to locate the commit timestamp for.
- * @param epoch
- * the epoch of the TSO server the current tso client is working with.
- * @param locator
- * a locator to find the commit timestamp in the system.
- * @return the commit timestamp joint with the location where it was found
- * or an object indicating that it was not found in the system
- * @throws IOException
- */
- public CommitTimestamp locateCellCommitTimestamp(long cellStartTimestamp, long epoch,
- CommitTimestampLocator locator) throws IOException {
-
- try {
- // 1) First check the cache
- Optional<Long> commitTimestamp = locator.readCommitTimestampFromCache(cellStartTimestamp);
- if (commitTimestamp.isPresent()) { // Valid commit timestamp
- return new CommitTimestamp(CACHE, commitTimestamp.get(), true);
- }
-
- // 2) Then check the commit table
- // If the data was written at a previous epoch, check whether the transaction was invalidated
- Optional<CommitTimestamp> commitTimeStamp = commitTableClient.getCommitTimestamp(cellStartTimestamp).get();
- if (commitTimeStamp.isPresent()) {
- return commitTimeStamp.get();
- }
-
- // 3) Read from shadow cell
- commitTimeStamp = readCommitTimestampFromShadowCell(cellStartTimestamp, locator);
- if (commitTimeStamp.isPresent()) {
- return commitTimeStamp.get();
- }
-
- // 4) Check the epoch and invalidate the entry
- // if the data was written by a transaction from a previous epoch (previous TSO)
- if (cellStartTimestamp < epoch) {
- boolean invalidated = commitTableClient.tryInvalidateTransaction(cellStartTimestamp).get();
- if (invalidated) { // Invalid commit timestamp
- return new CommitTimestamp(COMMIT_TABLE, CommitTable.INVALID_TRANSACTION_MARKER, false);
- }
- }
-
- // 5) We did not manage to invalidate the transactions then check the commit table
- commitTimeStamp = commitTableClient.getCommitTimestamp(cellStartTimestamp).get();
- if (commitTimeStamp.isPresent()) {
- return commitTimeStamp.get();
- }
-
- // 6) Read from shadow cell
- commitTimeStamp = readCommitTimestampFromShadowCell(cellStartTimestamp, locator);
- if (commitTimeStamp.isPresent()) {
- return commitTimeStamp.get();
- }
-
- // *) Otherwise return not found
- return new CommitTimestamp(NOT_PRESENT, -1L /** TODO Check if we should return this */, true);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new IOException("Interrupted while finding commit timestamp", e);
- } catch (ExecutionException e) {
- throw new IOException("Problem finding commit timestamp", e);
- }
-
- }
-
- /**
- * @see java.io.Closeable#close()
- */
- @Override
- public final void close() throws IOException {
-
- tsoClient.close();
- commitTableClient.close();
-
- }
-
- // ----------------------------------------------------------------------------------------------------------------
- // Helper methods
- // ----------------------------------------------------------------------------------------------------------------
-
- private void enforceTransactionIsInRunningState(Transaction transaction) {
-
- if (transaction.getStatus() != Status.RUNNING) {
- throw new IllegalArgumentException("Transaction was already " + transaction.getStatus());
- }
-
- }
-
- @SuppressWarnings("unchecked")
- // NOTE: We are sure that tx is not parametrized
- private AbstractTransaction<? extends CellId> enforceAbstractTransactionAsParam(Transaction tx) {
-
- if (tx instanceof AbstractTransaction) {
- return (AbstractTransaction<? extends CellId>) tx;
- } else {
- throw new IllegalArgumentException(
- "The transaction object passed is not an instance of AbstractTransaction");
- }
-
- }
-
- private void markReadOnlyTransaction(AbstractTransaction<? extends CellId> readOnlyTx) {
-
- readOnlyTx.setStatus(Status.COMMITTED_RO);
-
- }
-
- private void commitRegularTransaction(AbstractTransaction<? extends CellId> tx)
- throws RollbackException, TransactionException
- {
-
- try {
-
- long commitTs = tsoClient.commit(tx.getStartTimestamp(), tx.getWriteSet()).get();
- certifyCommitForTx(tx, commitTs);
- updateShadowCellsAndRemoveCommitTableEntry(tx, postCommitter);
-
- } catch (ExecutionException e) {
-
- if (e.getCause() instanceof AbortException) { // TSO reports Tx conflicts as AbortExceptions in the future
- rollback(tx);
- rolledbackTxsCounter.inc();
- throw new RollbackException("Conflicts detected in tx writeset", e.getCause());
- }
-
- if (e.getCause() instanceof ServiceUnavailableException
- ||
- e.getCause() instanceof NewTSOException
- ||
- e.getCause() instanceof ConnectionException) {
-
- errorTxsCounter.inc();
- try {
- LOG.warn("Can't contact the TSO for receiving outcome for Tx {}. Checking Commit Table...", tx);
- // Check the commit table to find if the target TSO woke up in the meantime and added the commit
- // TODO: Decide what we should we do if we can not contact the commit table
- Optional<CommitTimestamp> commitTimestamp =
- commitTableClient.getCommitTimestamp(tx.getStartTimestamp()).get();
- if (commitTimestamp.isPresent()) {
- if (commitTimestamp.get().isValid()) {
- LOG.warn("{}: Valid commit TS found in Commit Table. Committing Tx...", tx);
- certifyCommitForTx(tx, commitTimestamp.get().getValue());
- postCommitter.updateShadowCells(tx); // But do NOT remove transaction from commit table
- } else { // Probably another Tx in a new TSO Server invalidated this transaction
- LOG.warn("{}: Invalidated commit TS found in Commit Table. Rolling-back...", tx);
- rollback(tx);
- throw new RollbackException(tx + " invalidated by other Tx started", e.getCause());
- }
- } else {
- LOG.warn("{}: Trying to invalidate Tx proactively in Commit Table...", tx);
- boolean invalidated = commitTableClient.tryInvalidateTransaction(tx.getStartTimestamp()).get();
- if (invalidated) {
- LOG.warn("{}: Invalidated proactively in Commit Table. Rolling-back Tx...", tx);
- invalidatedTxsCounter.inc();
- rollback(tx); // Rollback proactively cause it's likely that a new TSOServer is now master
- throw new RollbackException(tx + " rolled-back precautionary", e.getCause());
- } else {
- LOG.warn("{}: Invalidation could NOT be completed. Re-checking Commit Table...", tx);
- // TODO: Decide what we should we do if we can not contact the commit table
- commitTimestamp = commitTableClient.getCommitTimestamp(tx.getStartTimestamp()).get();
- if (commitTimestamp.isPresent() && commitTimestamp.get().isValid()) {
- LOG.warn("{}: Valid commit TS found in Commit Table. Committing Tx...", tx);
- certifyCommitForTx(tx, commitTimestamp.get().getValue());
- postCommitter.updateShadowCells(tx); // But do NOT remove transaction from commit table
- } else {
- LOG.error("{}: Can't determine Transaction outcome", tx);
- throw new TransactionException(tx + ": cannot determine Tx outcome");
- }
- }
- }
- } catch (ExecutionException e1) {
- throw new TransactionException(tx + ": problem reading commitTS from Commit Table", e1);
- } catch (InterruptedException e1) {
- Thread.currentThread().interrupt();
- throw new TransactionException(tx + ": interrupted while reading commitTS from Commit Table", e1);
- }
- } else {
- throw new TransactionException(tx + ": cannot determine Tx outcome", e.getCause());
- }
- } catch (InterruptedException ie) {
- Thread.currentThread().interrupt();
- throw new TransactionException(tx + ": interrupted during commit", ie);
-
- }
-
- }
-
- private void updateShadowCellsAndRemoveCommitTableEntry(final AbstractTransaction<? extends CellId> tx,
- final PostCommitActions postCommitter) {
-
- Futures.transform(postCommitter.updateShadowCells(tx), new Function<Void, Void>() {
- @Override
- public Void apply(Void aVoid) {
- postCommitter.removeCommitTableEntry(tx);
- return null;
- }
- });
-
- }
-
- private void certifyCommitForTx(AbstractTransaction<? extends CellId> txToSetup, long commitTS) {
-
- txToSetup.setStatus(Status.COMMITTED);
- txToSetup.setCommitTimestamp(commitTS);
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/transaction-client/src/main/java/com/yahoo/omid/transaction/CommitTimestampLocator.java
----------------------------------------------------------------------
diff --git a/transaction-client/src/main/java/com/yahoo/omid/transaction/CommitTimestampLocator.java b/transaction-client/src/main/java/com/yahoo/omid/transaction/CommitTimestampLocator.java
deleted file mode 100644
index 13399c5..0000000
--- a/transaction-client/src/main/java/com/yahoo/omid/transaction/CommitTimestampLocator.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.google.common.base.Optional;
-
-import java.io.IOException;
-
-/**
- * An behavior that needs to be implemented by transaction managers
- * to try to locate the possible commit timestamp for a cell. It
- * allows to the transaction managers implementing this interface to
- * search the commit timestamp in a local cache of their own or in
- * their particular shadow cells implementation.
- */
-public interface CommitTimestampLocator {
-
- public Optional<Long> readCommitTimestampFromCache(long startTimestamp);
-
- public Optional<Long> readCommitTimestampFromShadowCell(long startTimestamp)
- throws IOException;
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/transaction-client/src/main/java/com/yahoo/omid/transaction/PostCommitActions.java
----------------------------------------------------------------------
diff --git a/transaction-client/src/main/java/com/yahoo/omid/transaction/PostCommitActions.java b/transaction-client/src/main/java/com/yahoo/omid/transaction/PostCommitActions.java
deleted file mode 100644
index 6b7b2ba..0000000
--- a/transaction-client/src/main/java/com/yahoo/omid/transaction/PostCommitActions.java
+++ /dev/null
@@ -1,41 +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.util.concurrent.ListenableFuture;
-import com.yahoo.omid.tso.client.CellId;
-
-public interface PostCommitActions {
-
- /**
- * Allows specific implementations to update the shadow cells.
- * @param transaction
- * the transaction to update shadow cells for
- * @return future signalling end of the computation
- */
- ListenableFuture<Void> updateShadowCells(AbstractTransaction<? extends CellId> transaction);
-
- /**
- * Allows specific implementations to remove the transaction entry from the commit table.
- * @param transaction
- * the transaction to remove the commit table entry
- * @return future signalling end of the computation
- */
- ListenableFuture<Void> removeCommitTableEntry(AbstractTransaction<? extends CellId> transaction);
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/transaction-client/src/main/java/com/yahoo/omid/transaction/RollbackException.java
----------------------------------------------------------------------
diff --git a/transaction-client/src/main/java/com/yahoo/omid/transaction/RollbackException.java b/transaction-client/src/main/java/com/yahoo/omid/transaction/RollbackException.java
deleted file mode 100644
index 1cd1176..0000000
--- a/transaction-client/src/main/java/com/yahoo/omid/transaction/RollbackException.java
+++ /dev/null
@@ -1,35 +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;
-
-public class RollbackException extends Exception {
-
- private static final long serialVersionUID = -9163407697376986830L;
-
- public RollbackException(String message) {
- super(message);
- }
-
- public RollbackException(String message, Throwable cause) {
- super(message, cause);
- }
-
- public RollbackException() {
- super();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/transaction-client/src/main/java/com/yahoo/omid/transaction/Transaction.java
----------------------------------------------------------------------
diff --git a/transaction-client/src/main/java/com/yahoo/omid/transaction/Transaction.java b/transaction-client/src/main/java/com/yahoo/omid/transaction/Transaction.java
deleted file mode 100644
index a10e2d3..0000000
--- a/transaction-client/src/main/java/com/yahoo/omid/transaction/Transaction.java
+++ /dev/null
@@ -1,76 +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.Optional;
-
-/**
- * This interface defines the transaction state & behavior exposed to users.
- */
-public interface Transaction {
-
- enum Status {
- RUNNING, COMMITTED, ROLLEDBACK, COMMITTED_RO
- }
-
- /**
- * Returns the transaction identifier
- * @return the transaction identifier
- */
- long getTransactionId();
-
- /**
- * Returns the epoch given by the TSOServer
- * @return the transaction's TSOServer epoch
- */
- long getEpoch();
-
- /**
- * Returns the current transaction {@link Status}
- * @return transaction status
- */
- Status getStatus();
-
- /**
- * Forces the transaction to rollback, even when there's an intention
- * to commit it.
- */
- void setRollbackOnly();
-
- /**
- * Returns whether the transaction was marked for rollback or not
- * @return whether the transaction is marked for rollback or not
- */
- boolean isRollbackOnly();
-
-
- /**
- * Set of methods to attach some metadata to a transaction object. One example
- * of such metadata are notifications
- *
- *
- * Expects they metadata stored under key "key" to be of the "Set" type,
- * append "value" to the existing set or creates a new one
- */
- void appendMetadata(String key, Object value);
-
- void setMetadata(String key, Object value);
-
- Optional<Object> getMetadata(String key);
-}
-
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/transaction-client/src/main/java/com/yahoo/omid/transaction/TransactionException.java
----------------------------------------------------------------------
diff --git a/transaction-client/src/main/java/com/yahoo/omid/transaction/TransactionException.java b/transaction-client/src/main/java/com/yahoo/omid/transaction/TransactionException.java
deleted file mode 100644
index f238a6b..0000000
--- a/transaction-client/src/main/java/com/yahoo/omid/transaction/TransactionException.java
+++ /dev/null
@@ -1,31 +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;
-
-public class TransactionException extends Exception {
-
- private static final long serialVersionUID = 7273525983622126275L;
-
- public TransactionException(String reason) {
- super(reason);
- }
-
- public TransactionException(String reason, Throwable e) {
- super(reason, e);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/transaction-client/src/main/java/com/yahoo/omid/transaction/TransactionManager.java
----------------------------------------------------------------------
diff --git a/transaction-client/src/main/java/com/yahoo/omid/transaction/TransactionManager.java b/transaction-client/src/main/java/com/yahoo/omid/transaction/TransactionManager.java
deleted file mode 100644
index d5782fa..0000000
--- a/transaction-client/src/main/java/com/yahoo/omid/transaction/TransactionManager.java
+++ /dev/null
@@ -1,66 +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 java.io.Closeable;
-
-/**
- * Provides the methods to manage transactions (create, commit...)
- */
-public interface TransactionManager extends Closeable {
-
- /**
- * Starts a new transaction.
- *
- * Creates & returns a {@link Transaction} interface implementation
- * that will be used in {@link TTable}'s methods for doing operations
- * on the transactional context defined by the returned object.
- *
- * @return transaction
- * representation of the created transaction
- * @throws TransactionException
- */
- public Transaction begin() throws TransactionException;
-
- /**
- * Commits a transaction.
- *
- * If the transaction was marked for rollback or has conflicts
- * with another concurrent transaction it will be rolledback
- * automatically and a {@link RollbackException} will be thrown.
- *
- * @param tx
- * transaction to be committed.
- * @throws RollbackException
- * thrown when transaction has conflicts with another transaction
- * or when was marked for rollback.
- * @throws TransactionException
- */
- public void commit(Transaction tx) throws RollbackException, TransactionException;
-
- /**
- * Aborts a transaction.
- *
- * Automatically rollbacks the changes performed by the transaction.
- *
- * @param tx
- * transaction to be rolled-back.
- */
- public void rollback(Transaction tx) throws TransactionException;
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/transaction-client/src/main/java/com/yahoo/omid/transaction/TransactionManagerException.java
----------------------------------------------------------------------
diff --git a/transaction-client/src/main/java/com/yahoo/omid/transaction/TransactionManagerException.java b/transaction-client/src/main/java/com/yahoo/omid/transaction/TransactionManagerException.java
deleted file mode 100644
index 09df8da..0000000
--- a/transaction-client/src/main/java/com/yahoo/omid/transaction/TransactionManagerException.java
+++ /dev/null
@@ -1,32 +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;
-
-public class TransactionManagerException extends Exception {
-
- private static final long serialVersionUID = 6811947817962260774L;
-
- public TransactionManagerException(String reason) {
- super(reason);
- }
-
- public TransactionManagerException(String reason, Throwable e) {
- super(reason, e);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/9cd856c6/transaction-client/src/main/java/com/yahoo/omid/tso/client/AbortException.java
----------------------------------------------------------------------
diff --git a/transaction-client/src/main/java/com/yahoo/omid/tso/client/AbortException.java b/transaction-client/src/main/java/com/yahoo/omid/tso/client/AbortException.java
deleted file mode 100644
index 670e87f..0000000
--- a/transaction-client/src/main/java/com/yahoo/omid/tso/client/AbortException.java
+++ /dev/null
@@ -1,27 +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.tso.client;
-
-/**
- * Thrown when the TSO server has aborted a transaction
- */
-public class AbortException extends Exception {
-
- private static final long serialVersionUID = 1861474360100681040L;
-
-}
\ No newline at end of file