test hserver support using file as metastore

Author: Time Hoo

app/src/test/java/io/hstream/testing/BasicExtension.java

@@ -16,6 +16,7 @@ public class BasicExtension implements BeforeEachCallback, AfterEachCallback {
 
   private static final Logger logger = LoggerFactory.getLogger(BasicExtension.class);
   private Path dataDir;
+  private Path metadataDir;
   private GenericContainer<?> zk;
   private GenericContainer<?> rq;
   private GenericContainer<?> hstore;
@@ -28,6 +29,7 @@ public void beforeEach(ExtensionContext context) throws Exception {
     printBeginFlag(context);
 
     dataDir = Files.createTempDirectory("hstream");
+    metadataDir = Files.createTempDirectory("hstream-meta");
 
     zk = makeZooKeeper();
     zk.start();
@@ -49,7 +51,8 @@ public void beforeEach(ExtensionContext context) throws Exception {
     options.port = hServerPort;
     options.internalPort = hServerInternalPort;
     options.metaHost = zkHost;
-    hserver = makeHServer(options, hServerAddress + ":" + hServerInternalPort, dataDir);
+    hserver =
+        makeHServer(options, hServerAddress + ":" + hServerInternalPort, dataDir, metadataDir);
     hserver.start();
     Thread.sleep(1000);
     Object testInstance = context.getRequiredTestInstance();

app/src/test/java/io/hstream/testing/ClusterExtension.java

@@ -27,6 +27,8 @@ public class ClusterExtension implements BeforeEachCallback, AfterEachCallback {
   private final List<String> hServerInnerUrls = new ArrayList<>(CLUSTER_SIZE);
   private String seedNodes;
   private Path dataDir;
+
+  private Path metadataDir;
   private GenericContainer<?> zk;
   private GenericContainer<?> rq;
   private GenericContainer<?> hstore;
@@ -43,6 +45,7 @@ public void beforeEach(ExtensionContext context) throws Exception {
     printBeginFlag(context);
 
     dataDir = Files.createTempDirectory("hstream");
+    metadataDir = Files.createTempDirectory("hstream-meta");
 
     TestUtils.SecurityOptions securityOptions = makeSecurityOptions(context.getTags());
 
@@ -76,7 +79,7 @@ public void beforeEach(ExtensionContext context) throws Exception {
       hServerInnerUrls.add(hServerAddress + ":" + hServerInnerPort);
     }
     seedNodes = hServerInnerUrls.stream().reduce((url1, url2) -> url1 + "," + url2).get();
-    hServers.addAll(bootstrapHServerCluster(hserverConfs, seedNodes, dataDir));
+    hServers.addAll(bootstrapHServerCluster(hserverConfs, seedNodes, dataDir, metadataDir));
     hServers.stream().forEach(h -> logger.info(h.getLogs()));
     Thread.sleep(3000);
 
@@ -115,6 +118,12 @@ public void beforeEach(ExtensionContext context) throws Exception {
                 .getClass()
                 .getMethod("setDataDir", Path.class)
                 .invoke(testInstance, dataDir));
+    silence(
+        () ->
+            testInstance
+                .getClass()
+                .getMethod("setMetadataDir", Path.class)
+                .invoke(testInstance, metadataDir));
     silence(
         () ->
             testInstance

app/src/test/java/io/hstream/testing/TestUtils.java

@@ -117,6 +117,10 @@ public static GenericContainer<?> makeRQLite() {
     return new GenericContainer<>(DockerImageName.parse("rqlite/rqlite")).withNetworkMode("host");
   }
 
+  public static GenericContainer<?> makeFile() {
+    return new GenericContainer<>(DockerImageName.parse("rqlite/rqlite")).withNetworkMode("host");
+  }
+
   private static DockerImageName getHStreamImageName() {
     String hstreamImageName = System.getenv("HSTREAM_IMAGE_NAME");
     if (hstreamImageName == null || hstreamImageName.equals("")) {
@@ -139,6 +143,9 @@ private static String getHStreamMetaStorePreference(String metaHost) {
     } else if (hstreamMetaStore.equalsIgnoreCase("RQLITE")) {
       logger.info("HSTREAM_META_STORE specified RQLITE as meta store");
       return "rq://" + metaHost + ":4001";
+    } else if (hstreamMetaStore.equalsIgnoreCase("FILE")) {
+      logger.info("HSTREAM_META_STORE specified FILE as meta store");
+      return "file:///data/metastore/hstream";
     } else {
       throw new RuntimeException("Invalid HSTREAM_META_STORE env variable value");
     }
@@ -184,10 +191,12 @@ public String toString() {
   }
 
   public static GenericContainer<?> makeHServer(
-      HServerCliOpts hserverConf, String seedNodes, Path dataDir) {
+      HServerCliOpts hserverConf, String seedNodes, Path dataDir, Path metadataDir) {
     return new GenericContainer<>(getHStreamImageName())
         .withNetworkMode("host")
         .withFileSystemBind(dataDir.toAbsolutePath().toString(), "/data/hstore", BindMode.READ_ONLY)
+        .withFileSystemBind(
+            metadataDir.toAbsolutePath().toString(), "/data/metastore", BindMode.READ_WRITE)
         .withFileSystemBind(hserverConf.securityOptions.dir, "/data/security", BindMode.READ_ONLY)
         .withCommand(
             "bash", "-c", " hstream-server" + hserverConf.toString() + " --seed-nodes " + seedNodes)
@@ -230,11 +239,11 @@ public String toString() {
   }
 
   public static List<GenericContainer<?>> bootstrapHServerCluster(
-      List<HServerCliOpts> hserverConfs, String seedNodes, Path dataDir)
+      List<HServerCliOpts> hserverConfs, String seedNodes, Path dataDir, Path metadataDir)
       throws IOException, InterruptedException {
     List<GenericContainer<?>> hServers = new ArrayList<>();
     for (HServerCliOpts hserverConf : hserverConfs) {
-      var hServer = makeHServer(hserverConf, seedNodes, dataDir);
+      var hServer = makeHServer(hserverConf, seedNodes, dataDir, metadataDir);
       hServers.add(hServer);
     }
     hServers.stream().parallel().forEach(GenericContainer::start);