Add INIT_WAIT_TIME CQL option to sleep after table creation

Signed-off-by: Boxuan Li <[email protected]>

docs/configs/janusgraph-cfg.md

@@ -435,6 +435,7 @@ CQL storage backend options
 | storage.cql.gc-grace-seconds | The number of seconds before tombstones (deletion markers) are eligible for garbage-collection. | Integer | (no default value) | FIXED |
 | storage.cql.heartbeat-interval | The connection heartbeat interval in milliseconds. | Long | (no default value) | MASKABLE |
 | storage.cql.heartbeat-timeout | How long the driver waits for the response (in milliseconds) to a heartbeat. | Long | (no default value) | MASKABLE |
+| storage.cql.init-wait-time | Number of milliseconds to sleep after creating each keyspace/table, only needed if table creation is asynchronous, e.g. Amazon Keyspace | Integer | (no default value) | LOCAL |
 | storage.cql.keyspace | The name of JanusGraph's keyspace.  It will be created if it does not exist. | String | janusgraph | LOCAL |
 | storage.cql.local-datacenter | The name of the local or closest Cassandra datacenter. This value will be passed into CqlSessionBuilder.withLocalDatacenter. | String | datacenter1 | MASKABLE |
 | storage.cql.local-max-connections-per-host | The maximum number of connections that can be created per host for local datacenter | Integer | 1 | MASKABLE |

docs/storage-backend/cassandra.md

@@ -334,6 +334,11 @@ storage.password=<your-service-password>
 storage.cql.keyspace=janusgraph
 storage.cql.local-datacenter=<your-datacenter, e.g. ap-east-1>
 
+# Wait for 30 seconds after each table creation, since
+# Amazon Keyspace creates tables asynchronously. Remember to
+# remove this option from config file after the creation of the graph.
+storage.cql.init-wait-time=30000
+
 # SSL related settings
 storage.cql.ssl.enabled=true
 storage.cql.ssl.truststore.location=<your-trust-store-location>

janusgraph-cql/src/main/java/org/janusgraph/diskstorage/cql/CQLConfigOptions.java

@@ -65,6 +65,13 @@ public interface CQLConfigOptions {
             ConfigOption.Type.MASKABLE,
             CQLStoreManager.CONSISTENCY_QUORUM);
 
+    ConfigOption<Integer> INIT_WAIT_TIME = new ConfigOption<>(
+            CQL_NS,
+            "init-wait-time",
+            "Number of milliseconds to sleep after creating each keyspace/table, only needed if table creation " +
+            "is asynchronous, e.g. Amazon Keyspace",
+            ConfigOption.Type.LOCAL, Integer.class);
+
     ConfigOption<Boolean> ONLY_USE_LOCAL_CONSISTENCY_FOR_SYSTEM_OPERATIONS =
         new ConfigOption<>(CQL_NS, "only-use-local-consistency-for-system-operations",
             "True to prevent any system queries from using QUORUM consistency " +

janusgraph-cql/src/main/java/org/janusgraph/diskstorage/cql/CQLKeyColumnValueStore.java

@@ -37,6 +37,7 @@
 import io.vavr.collection.Iterator;
 import io.vavr.control.Try;
 import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.janusgraph.core.JanusGraphException;
 import org.janusgraph.diskstorage.BackendException;
 import org.janusgraph.diskstorage.Entry;
 import org.janusgraph.diskstorage.EntryList;
@@ -86,6 +87,7 @@
 import static org.janusgraph.diskstorage.cql.CQLConfigOptions.COMPACTION_OPTIONS;
 import static org.janusgraph.diskstorage.cql.CQLConfigOptions.COMPACTION_STRATEGY;
 import static org.janusgraph.diskstorage.cql.CQLConfigOptions.GC_GRACE_SECONDS;
+import static org.janusgraph.diskstorage.cql.CQLConfigOptions.INIT_WAIT_TIME;
 import static org.janusgraph.diskstorage.cql.CQLConfigOptions.SPECULATIVE_RETRY;
 import static org.janusgraph.diskstorage.cql.CQLTransaction.getTransaction;
 import static org.janusgraph.graphdb.configuration.GraphDatabaseConfiguration.STORE_META_TIMESTAMPS;
@@ -187,6 +189,13 @@ public CQLKeyColumnValueStore(final CQLStoreManager storeManager, final String t
 
         if(shouldInitializeTable()) {
             initializeTable(this.session, this.storeManager.getKeyspaceName(), tableName, configuration);
+            if (configuration.has(INIT_WAIT_TIME) && configuration.get(INIT_WAIT_TIME) > 0) {
+                try {
+                    Thread.sleep(configuration.get(INIT_WAIT_TIME));
+                } catch (InterruptedException e) {
+                    throw new JanusGraphException("Interrupted while waiting for table initialization to complete", e);
+                }
+            }
         }
 
         if (this.storeManager.getFeatures().hasOrderedScan()) {

janusgraph-cql/src/main/java/org/janusgraph/diskstorage/cql/CQLStoreManager.java

@@ -28,6 +28,7 @@
 import io.vavr.collection.Iterator;
 import io.vavr.collection.Seq;
 import io.vavr.concurrent.Future;
+import org.janusgraph.core.JanusGraphException;
 import org.janusgraph.diskstorage.BackendException;
 import org.janusgraph.diskstorage.BaseTransactionConfig;
 import org.janusgraph.diskstorage.PermanentBackendException;
@@ -70,6 +71,7 @@
 import static org.janusgraph.diskstorage.cql.CQLConfigOptions.BACK_PRESSURE_CLASS;
 import static org.janusgraph.diskstorage.cql.CQLConfigOptions.BACK_PRESSURE_LIMIT;
 import static org.janusgraph.diskstorage.cql.CQLConfigOptions.EXECUTOR_SERVICE_MAX_SHUTDOWN_WAIT_TIME;
+import static org.janusgraph.diskstorage.cql.CQLConfigOptions.INIT_WAIT_TIME;
 import static org.janusgraph.diskstorage.cql.CQLConfigOptions.KEYSPACE;
 import static org.janusgraph.diskstorage.cql.CQLConfigOptions.NETWORK_TOPOLOGY_REPLICATION_STRATEGY;
 import static org.janusgraph.diskstorage.cql.CQLConfigOptions.REPLICATION_FACTOR;
@@ -232,6 +234,14 @@ void initializeKeyspace(){
             .ifNotExists()
             .withReplicationOptions(replication)
             .build());
+
+        if (configuration.has(INIT_WAIT_TIME) && configuration.get(INIT_WAIT_TIME) > 0) {
+            try {
+                Thread.sleep(configuration.get(INIT_WAIT_TIME));
+            } catch (InterruptedException e) {
+                throw new JanusGraphException("Interrupted while waiting for keyspace initialization to complete", e);
+            }
+        }
     }
 
     public ExecutorService getExecutorService() {

janusgraph-cql/src/test/java/org/janusgraph/diskstorage/cql/CQLConfigTest.java

@@ -65,6 +65,7 @@
 import static org.janusgraph.diskstorage.cql.CQLConfigOptions.EXECUTOR_SERVICE_MAX_SHUTDOWN_WAIT_TIME;
 import static org.janusgraph.diskstorage.cql.CQLConfigOptions.FILE_CONFIGURATION;
 import static org.janusgraph.diskstorage.cql.CQLConfigOptions.HEARTBEAT_TIMEOUT;
+import static org.janusgraph.diskstorage.cql.CQLConfigOptions.INIT_WAIT_TIME;
 import static org.janusgraph.diskstorage.cql.CQLConfigOptions.KEYSPACE;
 import static org.janusgraph.diskstorage.cql.CQLConfigOptions.LOCAL_DATACENTER;
 import static org.janusgraph.diskstorage.cql.CQLConfigOptions.METADATA_SCHEMA_ENABLED;
@@ -203,6 +204,14 @@ public void testStorageVersionSet() {
         mgmt.rollback();
     }
 
+    @Test
+    public void testInitWaitTime() {
+        WriteConfiguration wc = getConfiguration();
+        wc.set(ConfigElement.getPath(INIT_WAIT_TIME), 5000);
+        graph = (StandardJanusGraph) JanusGraphFactory.open(wc);
+        assertTrue(graph.isOpen());
+    }
+
     private String getCurrentPartitionerName() {
         switch (System.getProperty("cassandra.docker.partitioner")) {
             case "byteordered":