support hstream server cluster (#37)

Author: daleiz

.github/workflows/main.yml

@@ -22,7 +22,7 @@ jobs:
           args: '--set-exit-if-changed'
 
       - name: Prepare test env
-        run:  script/prepare-test-env.sh
+        run:  script/prepare-test-env-latest.sh
 
       - name: Set up JDK 11
         uses: actions/setup-java@v2

README.md

@@ -3,8 +3,8 @@
 ![Build Status](https://github.com/hstreamdb/hstreamdb-java/actions/workflows/main.yml/badge.svg)
 [![Maven Central](https://img.shields.io/maven-central/v/io.hstream/hstreamdb-java)](https://search.maven.org/artifact/io.hstream/hstreamdb-java)
 [![javadoc](https://javadoc.io/badge2/io.hstream/hstreamdb-java/0.5.1/javadoc.svg)](https://javadoc.io/doc/io.hstream/hstreamdb-java/0.5.1)
-[![Snapshot Artifacts](https://img.shields.io/nexus/s/https/s01.oss.sonatype.org/io.hstream/hstreamdb-java.svg)](https://s01.oss.sonatype.org/content/repositories/snapshots/io/hstream/hstreamdb-java/0.5.1-SNAPSHOT/)
-[![javadoc](https://javadoc.io/badge2/io.hstream/hstreamdb-java/0.5.1-SNAPSHOT/javadoc.svg)](https://hstreamdb.github.io/hstreamdb-java/javadoc/)
+[![Snapshot Artifacts](https://img.shields.io/nexus/s/https/s01.oss.sonatype.org/io.hstream/hstreamdb-java.svg)](https://s01.oss.sonatype.org/content/repositories/snapshots/io/hstream/hstreamdb-java/0.6.0-SNAPSHOT/)
+[![javadoc](https://javadoc.io/badge2/io.hstream/hstreamdb-java/0.6.0-SNAPSHOT/javadoc.svg)](https://hstreamdb.github.io/hstreamdb-java/javadoc/)
 
 This is the offical Java client library for [HStreamDB](https://hstream.io/).
 

client/build.gradle

@@ -8,7 +8,7 @@ plugins {
 }
 
 group = 'io.hstream'
-version = '0.5.1'
+version = '0.6.0-SNAPSHOT'
 
 repositories {
     mavenCentral()

client/src/main/java/io/hstream/impl/ChannelProvider.java

@@ -0,0 +1,32 @@
+package io.hstream.impl;
+
+import io.grpc.ManagedChannel;
+import io.grpc.ManagedChannelBuilder;
+import java.io.Closeable;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class ChannelProvider implements Closeable {
+
+  private static final int DEFAULT_CHANNEL_PROVIDER_SIZE = 64;
+
+  private final ConcurrentHashMap<String, ManagedChannel> provider;
+
+  public ChannelProvider(int size) {
+    provider = new ConcurrentHashMap<>(size);
+  }
+
+  public ChannelProvider() {
+    this(DEFAULT_CHANNEL_PROVIDER_SIZE);
+  }
+
+  public ManagedChannel get(String serverUrl) {
+    return provider.computeIfAbsent(
+        serverUrl, url -> ManagedChannelBuilder.forTarget(url).usePlaintext().build());
+  }
+
+  @Override
+  public void close() {
+    provider.forEachValue(Long.MAX_VALUE, managedChannel -> managedChannel.shutdown());
+    provider.clear();
+  }
+}

client/src/main/java/io/hstream/impl/ConsumerBuilderImpl.java

@@ -6,22 +6,21 @@
 import io.hstream.ConsumerBuilder;
 import io.hstream.HRecordReceiver;
 import io.hstream.RawRecordReceiver;
-import io.hstream.internal.HStreamApiGrpc;
+import java.util.List;
 
 public class ConsumerBuilderImpl implements ConsumerBuilder {
 
-  private HStreamApiGrpc.HStreamApiStub grpcStub;
-  private HStreamApiGrpc.HStreamApiBlockingStub grpcBlockingStub;
+  private final List<String> serverUrls;
+  private final ChannelProvider channelProvider;
+
   private String name;
   private String subscription;
   private RawRecordReceiver rawRecordReceiver;
   private HRecordReceiver hRecordReceiver;
 
-  public ConsumerBuilderImpl(
-      HStreamApiGrpc.HStreamApiStub grpcStub,
-      HStreamApiGrpc.HStreamApiBlockingStub grpcBlockingStub) {
-    this.grpcStub = grpcStub;
-    this.grpcBlockingStub = grpcBlockingStub;
+  public ConsumerBuilderImpl(List<String> serverUrls, ChannelProvider channelProvider) {
+    this.serverUrls = serverUrls;
+    this.channelProvider = channelProvider;
   }
 
   @Override
@@ -50,11 +49,9 @@ public ConsumerBuilder hRecordReceiver(HRecordReceiver hRecordReceiver) {
 
   @Override
   public Consumer build() {
-    checkNotNull(grpcStub);
-    checkNotNull(grpcBlockingStub);
     checkNotNull(subscription);
     checkState(rawRecordReceiver != null || hRecordReceiver != null);
     return new ConsumerImpl(
-        grpcStub, grpcBlockingStub, name, subscription, rawRecordReceiver, hRecordReceiver);
+        serverUrls, channelProvider, name, subscription, rawRecordReceiver, hRecordReceiver);
   }
 }

client/src/main/java/io/hstream/impl/ConsumerImpl.java

@@ -3,6 +3,7 @@
 import com.google.common.util.concurrent.AbstractService;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.google.protobuf.InvalidProtocolBufferException;
+import io.grpc.ManagedChannelBuilder;
 import io.grpc.stub.StreamObserver;
 import io.hstream.*;
 import io.hstream.internal.*;
@@ -18,29 +19,34 @@
 public class ConsumerImpl extends AbstractService implements Consumer {
   private static final Logger logger = LoggerFactory.getLogger(ConsumerImpl.class);
 
-  private HStreamApiGrpc.HStreamApiStub grpcStub;
-  private HStreamApiGrpc.HStreamApiBlockingStub grpcBlockingStub;
-  private String consumerName;
-  private String subscriptionId;
-  private RawRecordReceiver rawRecordReceiver;
-  private HRecordReceiver hRecordReceiver;
+  private final List<String> serverUrls;
+  private final ChannelProvider channelProvider;
+
+  private final String consumerName;
+  private final String subscriptionId;
+  private final RawRecordReceiver rawRecordReceiver;
+  private final HRecordReceiver hRecordReceiver;
 
   private ExecutorService executorService;
 
+  private HStreamApiGrpc.HStreamApiStub fetchStub;
+
   private final StreamObserver<StreamingFetchResponse> responseStream;
   private final StreamObserver<StreamingFetchRequest> requestStream;
 
-  private final AtomicBoolean inited = new AtomicBoolean(false);
+  private final AtomicBoolean isInitialized = new AtomicBoolean(false);
 
   public ConsumerImpl(
-      HStreamApiGrpc.HStreamApiStub grpcStub,
-      HStreamApiGrpc.HStreamApiBlockingStub grpcBlockingStub,
+      List<String> serverUrls,
+      ChannelProvider channelProvider,
       String consumerName,
       String subscriptionId,
       RawRecordReceiver rawRecordReceiver,
       HRecordReceiver hRecordReceiver) {
-    this.grpcStub = grpcStub;
-    this.grpcBlockingStub = grpcBlockingStub;
+
+    this.serverUrls = serverUrls;
+    this.channelProvider = channelProvider;
+
     if (consumerName == null) {
       this.consumerName = UUID.randomUUID().toString();
     } else {
@@ -54,11 +60,12 @@ public ConsumerImpl(
         Executors.newSingleThreadExecutor(
             new ThreadFactoryBuilder().setNameFormat("receiver-running-pool-%d").build());
 
+    fetchStub = createFetchStub();
     responseStream =
         new StreamObserver<StreamingFetchResponse>() {
           @Override
           public void onNext(StreamingFetchResponse value) {
-            if (inited.compareAndSet(false, true)) {
+            if (isInitialized.compareAndSet(false, true)) {
               // notifyStarted();
             }
 
@@ -102,7 +109,7 @@ public void onNext(StreamingFetchResponse value) {
 
           @Override
           public void onError(Throwable t) {
-            if (inited.compareAndSet(false, true)) {
+            if (isInitialized.compareAndSet(false, true)) {
               logger.error(
                   "consumer {} attach to subscription {} error: {}",
                   ConsumerImpl.this.consumerName,
@@ -123,7 +130,19 @@ public void onError(Throwable t) {
           public void onCompleted() {}
         };
 
-    this.requestStream = grpcStub.streamingFetch(responseStream);
+    this.requestStream = fetchStub.streamingFetch(responseStream);
+  }
+
+  private HStreamApiGrpc.HStreamApiStub createFetchStub() {
+    ServerNode serverNode =
+        HStreamApiGrpc.newBlockingStub(
+                ManagedChannelBuilder.forTarget(serverUrls.get(0)).usePlaintext().build())
+            .lookupSubscription(
+                LookupSubscriptionRequest.newBuilder().setSubscriptionId(subscriptionId).build())
+            .getServerNode();
+
+    String serverUrl = serverNode.getHost() + ":" + serverNode.getPort();
+    return HStreamApiGrpc.newStub(channelProvider.get(serverUrl));
   }
 
   @Override

client/src/main/java/io/hstream/impl/HStreamClientBuilderImpl.java

@@ -4,6 +4,7 @@
 
 import io.hstream.HStreamClient;
 import io.hstream.HStreamClientBuilder;
+import java.util.List;
 
 public class HStreamClientBuilderImpl implements HStreamClientBuilder {
 
@@ -18,6 +19,11 @@ public HStreamClientBuilder serviceUrl(String serviceUrl) {
   @Override
   public HStreamClient build() {
     checkNotNull(serviceUrl);
-    return new HStreamClientImpl(serviceUrl);
+    List<String> serverUrls = parseServerUrls(serviceUrl);
+    return new HStreamClientImpl(serverUrls);
+  }
+
+  private List<String> parseServerUrls(String url) {
+    return List.of(url.strip().split(","));
   }
 }

client/src/main/java/io/hstream/impl/HStreamClientImpl.java

@@ -2,13 +2,12 @@
 
 import com.google.protobuf.Empty;
 import io.grpc.ManagedChannel;
-import io.grpc.ManagedChannelBuilder;
 import io.hstream.*;
-import io.hstream.internal.DeleteStreamRequest;
-import io.hstream.internal.DeleteSubscriptionRequest;
-import io.hstream.internal.HStreamApiGrpc;
-import io.hstream.internal.ListStreamsResponse;
+import io.hstream.Stream;
+import io.hstream.Subscription;
+import io.hstream.internal.*;
 import io.hstream.util.GrpcUtils;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
@@ -19,32 +18,44 @@ public class HStreamClientImpl implements HStreamClient {
 
   private static final Logger logger = LoggerFactory.getLogger(HStreamClientImpl.class);
 
-  private final ManagedChannel managedChannel;
-  private final HStreamApiGrpc.HStreamApiStub stub;
-  private final HStreamApiGrpc.HStreamApiBlockingStub blockingStub;
-
   private static final short DEFAULT_STREAM_REPLICATOR = 3;
 
-  public HStreamClientImpl(String serviceUrl) {
-    ManagedChannel channel = ManagedChannelBuilder.forTarget(serviceUrl).usePlaintext().build();
-    this.managedChannel = channel;
-    this.stub = HStreamApiGrpc.newStub(channel);
-    this.blockingStub = HStreamApiGrpc.newBlockingStub(channel);
+  private final ChannelProvider channelProvider;
+  private final List<String> bootstrapServerUrls;
+  private final List<String> initializedServerUrls;
+
+  private HStreamApiGrpc.HStreamApiBlockingStub blockingStub;
+
+  public HStreamClientImpl(List<String> bootstrapServerUrls) {
+    this.bootstrapServerUrls = bootstrapServerUrls;
+    channelProvider = new ChannelProvider();
+
+    ManagedChannel channel = channelProvider.get(bootstrapServerUrls.get(0));
+    blockingStub = HStreamApiGrpc.newBlockingStub(channel);
+    DescribeClusterResponse describeClusterResponse =
+        blockingStub.describeCluster(Empty.newBuilder().build());
+    List<ServerNode> serverNodes = describeClusterResponse.getServerNodesList();
+    initializedServerUrls = new ArrayList<>(serverNodes.size());
+    for (ServerNode serverNode : serverNodes) {
+      String host = serverNode.getHost();
+      int port = serverNode.getPort();
+      initializedServerUrls.add(host + ":" + port);
+    }
   }
 
   @Override
   public ProducerBuilder newProducer() {
-    return new ProducerBuilderImpl(stub);
+    return new ProducerBuilderImpl(initializedServerUrls, channelProvider);
   }
 
   @Override
   public ConsumerBuilder newConsumer() {
-    return new ConsumerBuilderImpl(stub, blockingStub);
+    return new ConsumerBuilderImpl(initializedServerUrls, channelProvider);
   }
 
   @Override
   public QueryerBuilder newQueryer() {
-    return new QueryerBuilderImpl(this, stub);
+    return new QueryerBuilderImpl(this, initializedServerUrls, channelProvider);
   }
 
   @Override
@@ -101,7 +112,7 @@ public void deleteSubscription(String subscriptionId) {
   }
 
   @Override
-  public void close() throws Exception {
-    managedChannel.shutdownNow();
+  public void close() {
+    channelProvider.close();
   }
 }

client/src/main/java/io/hstream/impl/ProducerBuilderImpl.java

@@ -4,20 +4,22 @@
 
 import io.hstream.Producer;
 import io.hstream.ProducerBuilder;
-import io.hstream.internal.HStreamApiGrpc;
+import java.util.List;
 
 public class ProducerBuilderImpl implements ProducerBuilder {
 
-  private HStreamApiGrpc.HStreamApiStub grpcStub;
-
   private String streamName;
 
   private boolean enableBatch = false;
 
   private int recordCountLimit = 1;
 
-  public ProducerBuilderImpl(HStreamApiGrpc.HStreamApiStub stub) {
-    this.grpcStub = stub;
+  private final List<String> serverUrls;
+  private final ChannelProvider channelProvider;
+
+  public ProducerBuilderImpl(List<String> serverUrls, ChannelProvider channelProvider) {
+    this.serverUrls = serverUrls;
+    this.channelProvider = channelProvider;
   }
 
   @Override
@@ -40,8 +42,7 @@ public ProducerBuilder recordCountLimit(int recordCountLimit) {
 
   @Override
   public Producer build() {
-    checkNotNull(grpcStub);
     checkNotNull(streamName);
-    return new ProducerImpl(grpcStub, streamName, enableBatch, recordCountLimit);
+    return new ProducerImpl(serverUrls, channelProvider, streamName, enableBatch, recordCountLimit);
   }
 }