[OMID-70] Adding world base timestamp generator. The

 generator maintains monotonicity and persists the maximal timestamp in each
 interval. The motivation for this commit is the integration with Apache
 Phoenix. Phoenix's customers requires mapping from timestamp to world time in
 order to debug their logs. This commit also contains augmentation of the
 proto in order to partially serialize transaction. This is also for Phoenix
 usage.

common/src/main/proto/TSOProto.proto

@@ -64,3 +64,4 @@ message HandshakeResponse {
     optional bool clientCompatible = 1;
     optional Capabilities serverCapabilities = 2;
 }
+

hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransaction.java

@@ -29,7 +29,7 @@
 public class HBaseTransaction extends AbstractTransaction<HBaseCellId> {
     private static final Logger LOG = LoggerFactory.getLogger(HBaseTransaction.class);
 
-    HBaseTransaction(long transactionId, long epoch, Set<HBaseCellId> writeSet, AbstractTransactionManager tm) {
+    public HBaseTransaction(long transactionId, long epoch, Set<HBaseCellId> writeSet, AbstractTransactionManager tm) {
         super(transactionId, epoch, writeSet, tm);
     }
 

tso-server/src/main/java/org/apache/omid/tso/TSOModule.java

@@ -24,6 +24,9 @@
 
 import javax.inject.Named;
 import javax.inject.Singleton;
+
+import org.apache.omid.tso.TSOServerConfig.TIMESTAMP_TYPE;
+
 import java.net.SocketException;
 import java.net.UnknownHostException;
 
@@ -43,7 +46,13 @@ protected void configure() {
 
         bind(TSOChannelHandler.class).in(Singleton.class);
         bind(TSOStateManager.class).to(TSOStateManagerImpl.class).in(Singleton.class);
-        bind(TimestampOracle.class).to(TimestampOracleImpl.class).in(Singleton.class);
+
+        if (config.getTimestampTypeEnum() == TIMESTAMP_TYPE.WORLD_TIME) {
+            bind(TimestampOracle.class).to(WorldClockOracleImpl.class).in(Singleton.class);
+        } else {
+            bind(TimestampOracle.class).to(TimestampOracleImpl.class).in(Singleton.class);
+        }
+
         bind(Panicker.class).to(SystemExitPanicker.class).in(Singleton.class);
 
         install(new BatchPoolModule(config));

tso-server/src/main/java/org/apache/omid/tso/TSOServerConfig.java

@@ -44,6 +44,11 @@ public static enum WAIT_STRATEGY {
         LOW_CPU
     };
 
+    public static enum TIMESTAMP_TYPE {
+      INCREMENTAL,
+      WORLD_TIME
+    };
+
     // ----------------------------------------------------------------------------------------------------------------
     // Instantiation
     // ----------------------------------------------------------------------------------------------------------------
@@ -82,6 +87,8 @@ public TSOServerConfig() {
 
     private String networkIfaceName = NetworkUtils.getDefaultNetworkInterface();
 
+    private String timestampType;
+
     public int getPort() {
         return port;
     }
@@ -130,6 +137,18 @@ public void setNetworkIfaceName(String networkIfaceName) {
         this.networkIfaceName = networkIfaceName;
     }
 
+    public String getTimestampType() {
+        return timestampType;
+    }
+
+    public void setTimestampType(String type) {
+        this.timestampType = type;
+    }
+
+    public TIMESTAMP_TYPE getTimestampTypeEnum() {
+        return TSOServerConfig.TIMESTAMP_TYPE.valueOf(timestampType);
+    }
+
     public Module getTimestampStoreModule() {
         return timestampStoreModule;
     }

tso-server/src/main/java/org/apache/omid/tso/WorldClockOracleImpl.java

@@ -0,0 +1,182 @@
+/*
+ * 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.tso;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+import org.apache.omid.metrics.Gauge;
+import org.apache.omid.metrics.MetricsRegistry;
+import org.apache.omid.timestamp.storage.TimestampStorage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.inject.Inject;
+import javax.inject.Singleton;
+
+import java.io.IOException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.omid.metrics.MetricsUtils.name;
+
+/**
+ * The Timestamp Oracle that gives monotonically increasing timestamps based on world time
+ */
+@Singleton
+public class WorldClockOracleImpl implements TimestampOracle {
+
+    private static final Logger LOG = LoggerFactory.getLogger(WorldClockOracleImpl.class);
+
+    static final long MAX_TX_PER_MS = 1_000_000; // 1 million
+    static final long TIMESTAMP_INTERVAL_MS = 10_000; // 10 seconds interval
+    private static final long TIMESTAMP_ALLOCATION_INTERVAL_MS = 7_000; // 7 seconds
+
+    private long lastTimestamp;
+    private long maxTimestamp;
+
+    private TimestampStorage storage;
+    private Panicker panicker;
+
+    private volatile long maxAllocatedTime;
+
+    private final ScheduledExecutorService scheduler =
+            Executors.newScheduledThreadPool(1, new ThreadFactoryBuilder().setNameFormat("ts-persist-%d").build());
+
+    private Runnable allocateTimestampsBatchTask;
+
+    private class AllocateTimestampBatchTask implements Runnable {
+        long previousMaxTime;
+
+        AllocateTimestampBatchTask(long previousMaxTime) {
+            this.previousMaxTime = previousMaxTime;
+        }
+
+        @Override
+        public void run() {
+            long newMaxTime = (System.currentTimeMillis() + TIMESTAMP_INTERVAL_MS) * MAX_TX_PER_MS;
+            try {
+                storage.updateMaxTimestamp(previousMaxTime, newMaxTime);
+                maxAllocatedTime = newMaxTime;
+                previousMaxTime = newMaxTime;
+            } catch (Throwable e) {
+                panicker.panic("Can't store the new max timestamp", e);
+            }
+        }
+    }
+
+    @Inject
+    public WorldClockOracleImpl(MetricsRegistry metrics,
+                               TimestampStorage tsStorage,
+                               Panicker panicker) throws IOException {
+
+        this.storage = tsStorage;
+        this.panicker = panicker;
+
+        metrics.gauge(name("tso", "maxTimestamp"), new Gauge<Long>() {
+            @Override
+            public Long getValue() {
+                return maxTimestamp;
+            }
+        });
+
+    }
+
+    @Override
+    public void initialize() throws IOException {
+
+        this.lastTimestamp = this.maxTimestamp = storage.getMaxTimestamp();
+
+        this.allocateTimestampsBatchTask = new AllocateTimestampBatchTask(lastTimestamp);
+
+        // Trigger first allocation of timestamps
+        scheduler.schedule(allocateTimestampsBatchTask, 0, TimeUnit.MILLISECONDS);
+
+        // Waiting for the current epoch to start. Occurs in case of failover when the previous TSO allocated the current time frame.
+        while ((System.currentTimeMillis() * MAX_TX_PER_MS) < this.lastTimestamp) {
+            try {
+                Thread.sleep(1000);
+            } catch (InterruptedException e) {
+               continue;
+            }
+        }
+
+        // Launch the periodic timestamp interval allocation. In this case, the timestamp interval is extended even though the TSO is idle.
+        // Because we are world time based, this guarantees that the first request after a long time does not need to wait for new interval allocation.
+        scheduler.scheduleAtFixedRate(allocateTimestampsBatchTask, TIMESTAMP_ALLOCATION_INTERVAL_MS, TIMESTAMP_ALLOCATION_INTERVAL_MS, TimeUnit.MILLISECONDS);
+    }
+
+    /**
+     * Returns the next timestamp if available. Otherwise spins till the ts-persist thread allocates a new timestamp.
+     */
+    @Override
+    public long next() {
+
+        long currentMsFirstTimestamp = System.currentTimeMillis() * MAX_TX_PER_MS;
+
+        // Return the next timestamp in case we are still in the same millisecond as the previous timestamp was. 
+        if (++lastTimestamp >= currentMsFirstTimestamp) {
+            return lastTimestamp;
+        }
+
+        if (currentMsFirstTimestamp >= maxTimestamp) { // Intentional race to reduce synchronization overhead in every access to maxTimestamp                                                                                                                       
+            while (maxAllocatedTime <= currentMsFirstTimestamp) { // Waiting for the interval allocation
+                try {
+                    Thread.sleep(1000);
+                } catch (InterruptedException e) {
+                   continue;
+                }
+            }
+            assert (maxAllocatedTime > maxTimestamp);
+            maxTimestamp = maxAllocatedTime;
+        }
+
+        lastTimestamp = currentMsFirstTimestamp;
+
+        return lastTimestamp;
+    }
+
+    @Override
+    public long getLast() {
+        return lastTimestamp;
+    }
+
+    @Override
+    public String toString() {
+        return String.format("TimestampOracle -> LastTimestamp: %d, MaxTimestamp: %d", lastTimestamp, maxTimestamp);
+    }
+
+    @VisibleForTesting
+    static class InMemoryTimestampStorage implements TimestampStorage {
+
+        long maxTime = 0;
+
+        @Override
+        public void updateMaxTimestamp(long previousMaxTime, long nextMaxTime) {
+            maxTime = nextMaxTime;
+            LOG.info("Updating max timestamp: (previous:{}, new:{})", previousMaxTime, nextMaxTime);
+        }
+
+        @Override
+        public long getMaxTimestamp() {
+            return maxTime;
+        }
+
+    }
+}

tso-server/src/main/resources/default-omid-server-configuration.yml

@@ -26,6 +26,10 @@ numConcurrentCTWriters: 2
 batchSizePerCTWriter: 25
 # When this timeout expires, the contents of the batch are flushed to the datastore
 batchPersistTimeoutInMs: 10
+# Timestamp generation strategy
+# INCREMENTAL - [Default] regular counter
+# WORLD_TIME - world time based counter
+timestampType: INCREMENTAL
 
 # Default module configuration (No TSO High Availability & in-memory storage for timestamp and commit tables)
 timestampStoreModule: !!org.apache.omid.tso.InMemoryTimestampStorageModule [ ]

tso-server/src/test/java/org/apache/omid/tso/TestWorldTimeOracle.java

@@ -0,0 +1,117 @@
+/*
+ * 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.tso;
+
+import org.apache.omid.metrics.MetricsRegistry;
+import org.apache.omid.timestamp.storage.TimestampStorage;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.io.IOException;
+import java.util.concurrent.CountDownLatch;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.verify;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+
+public class TestWorldTimeOracle {
+
+    private static final Logger LOG = LoggerFactory.getLogger(TestWorldTimeOracle.class);
+
+    @Mock
+    private MetricsRegistry metrics;
+    @Mock
+    private Panicker panicker;
+    @Mock
+    private TimestampStorage timestampStorage;
+
+    // Component under test
+    @InjectMocks
+    private WorldClockOracleImpl worldClockOracle;
+
+    @BeforeMethod(alwaysRun = true, timeOut = 30_000)
+    public void initMocksAndComponents() {
+        MockitoAnnotations.initMocks(this);
+    }
+
+    @Test(timeOut = 30_000)
+    public void testMonotonicTimestampGrowth() throws Exception {
+
+        // Intialize component under test
+        worldClockOracle.initialize();
+
+        long last = worldClockOracle.next();
+
+        int timestampIntervalSec = (int) (WorldClockOracleImpl.TIMESTAMP_INTERVAL_MS / 1000) * 2;
+        for (int i = 0; i < timestampIntervalSec; i++) {
+            long current = worldClockOracle.next();
+            assertTrue(current > last+1 , "Timestamp should be based on world time");
+            last = current;
+            Thread.sleep(1000);
+        }
+
+        assertTrue(worldClockOracle.getLast() == last);
+        LOG.info("Last timestamp: {}", last);
+    }
+
+    @Test(timeOut = 10_000)
+    public void testTimestampOraclePanicsWhenTheStorageHasProblems() throws Exception {
+
+        // Intialize component under test
+        worldClockOracle.initialize();
+
+        // Cause an exception when updating the max timestamp
+        final CountDownLatch updateMaxTimestampMethodCalled = new CountDownLatch(1);
+        doAnswer(new Answer() {
+            @Override
+            public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
+                updateMaxTimestampMethodCalled.countDown();
+                throw new RuntimeException("Out of memory or something");
+            }
+        }).when(timestampStorage).updateMaxTimestamp(anyLong(), anyLong());
+
+        // Make the previous exception to be thrown
+        Thread allocThread = new Thread("AllocThread") {
+            @Override
+            public void run() {
+                while (true) {
+                    worldClockOracle.next();
+                }
+            }
+        };
+        allocThread.start();
+
+        updateMaxTimestampMethodCalled.await();
+
+        // Verify that it has blown up
+        verify(panicker, atLeastOnce()).panic(anyString(), any(Throwable.class));
+    }
+
+}