Notify fixes (#863)

- Change notify reconnect to use a backoff - Change unlisten to active on dispose of EmitterProcessor instead of longer processing chain - Fix Hibernate trying to connect on startup to check if embedded database - Fix incorrectly using EmitterProcessor - Fix invalid json causing a reconnect Signed-off-by: Steven Sheehy <[email protected]>
hiero-ledger · Jul 12, 2020 · 6a43e2d · 6a43e2d
1 parent b78d15c
commit 6a43e2d
Show file tree

Hide file tree

Showing 25 changed files with 32 additions and 26 deletions.
diff --git a/charts/hedera-mirror-grpc/Chart.yaml b/charts/hedera-mirror-grpc/Chart.yaml
@@ -1,5 +1,5 @@
 apiVersion: v1
-appVersion: 0.15.0-rc1
+appVersion: 0.15.0-rc2
 description: GRPC API for the Hedera Mirror Node
 home: https://github.com/hashgraph/hedera-mirror-node
 icon: https://camo.githubusercontent.com/cca6b767847bb8ca5c7059481ba13a5fc81c5938/68747470733a2f2f7777772e6865646572612e636f6d2f6c6f676f2d6361706974616c2d686261722d776f72646d61726b2e6a7067

diff --git a/charts/hedera-mirror-importer/Chart.yaml b/charts/hedera-mirror-importer/Chart.yaml
@@ -1,5 +1,5 @@
 apiVersion: v1
-appVersion: 0.15.0-rc1
+appVersion: 0.15.0-rc2
 description: Hedera Mirror Importer downloads and validates file streams from the cloud and imports them into the database
 home: https://github.com/hashgraph/hedera-mirror-node
 icon: https://camo.githubusercontent.com/cca6b767847bb8ca5c7059481ba13a5fc81c5938/68747470733a2f2f7777772e6865646572612e636f6d2f6c6f676f2d6361706974616c2d686261722d776f72646d61726b2e6a7067

diff --git a/charts/hedera-mirror-rest/Chart.yaml b/charts/hedera-mirror-rest/Chart.yaml
@@ -1,5 +1,5 @@
 apiVersion: v1
-appVersion: 0.15.0-rc1
+appVersion: 0.15.0-rc2
 description: REST API for the Hedera Mirror Node
 home: https://github.com/hashgraph/hedera-mirror-node
 icon: https://camo.githubusercontent.com/cca6b767847bb8ca5c7059481ba13a5fc81c5938/68747470733a2f2f7777772e6865646572612e636f6d2f6c6f676f2d6361706974616c2d686261722d776f72646d61726b2e6a7067

diff --git a/charts/hedera-mirror/Chart.yaml b/charts/hedera-mirror/Chart.yaml
@@ -1,5 +1,5 @@
 apiVersion: v1
-appVersion: 0.15.0-rc1
+appVersion: 0.15.0-rc2
 description: Hedera Mirror Node mirrors data from Hedera nodes and serves it via an API
 home: https://github.com/hashgraph/hedera-mirror-node
 icon: https://camo.githubusercontent.com/cca6b767847bb8ca5c7059481ba13a5fc81c5938/68747470733a2f2f7777772e6865646572612e636f6d2f6c6f676f2d6361706974616c2d686261722d776f72646d61726b2e6a7067

diff --git a/charts/hedera-mirror/charts/hedera-mirror-grpc-0.3.0.tgz b/charts/hedera-mirror/charts/hedera-mirror-grpc-0.3.0.tgz
diff --git a/charts/hedera-mirror/charts/hedera-mirror-importer-0.3.0.tgz b/charts/hedera-mirror/charts/hedera-mirror-importer-0.3.0.tgz
diff --git a/charts/hedera-mirror/charts/hedera-mirror-rest-0.3.0.tgz b/charts/hedera-mirror/charts/hedera-mirror-rest-0.3.0.tgz
diff --git a/docker-compose.yml b/docker-compose.yml
@@ -17,7 +17,7 @@ services:
       - 5432:5432
 
   grpc:
-    image: gcr.io/mirrornode/hedera-mirror-grpc:0.15.0-rc1
+    image: gcr.io/mirrornode/hedera-mirror-grpc:0.15.0-rc2
     restart: unless-stopped
     environment:
       HEDERA_MIRROR_GRPC_DB_HOST: db
@@ -28,7 +28,7 @@ services:
       - 5600:5600
 
   importer:
-    image: gcr.io/mirrornode/hedera-mirror-importer:0.15.0-rc1
+    image: gcr.io/mirrornode/hedera-mirror-importer:0.15.0-rc2
     restart: unless-stopped
     environment:
       HEDERA_MIRROR_IMPORTER_DATAPATH: /var/lib/hedera-mirror-importer
@@ -39,7 +39,7 @@ services:
       - ./application.yml:/usr/etc/hedera-mirror-importer/application.yml
 
   rest:
-    image: gcr.io/mirrornode/hedera-mirror-rest:0.15.0-rc1
+    image: gcr.io/mirrornode/hedera-mirror-rest:0.15.0-rc2
     environment:
       HEDERA_MIRROR_REST_DB_HOST: db
     restart: unless-stopped

diff --git a/hedera-mirror-coverage/pom.xml b/hedera-mirror-coverage/pom.xml
@@ -11,7 +11,7 @@
     <parent>
         <groupId>com.hedera</groupId>
         <artifactId>hedera-mirror-node</artifactId>
-        <version>0.15.0-rc1</version>
+        <version>0.15.0-rc2</version>
     </parent>
 
     <dependencies>

diff --git a/hedera-mirror-datagenerator/pom.xml b/hedera-mirror-datagenerator/pom.xml
@@ -11,7 +11,7 @@
     <parent>
         <groupId>com.hedera</groupId>
         <artifactId>hedera-mirror-node</artifactId>
-        <version>0.15.0-rc1</version>
+        <version>0.15.0-rc2</version>
     </parent>
 
     <properties>

diff --git a/hedera-mirror-grpc/pom.xml b/hedera-mirror-grpc/pom.xml
@@ -11,7 +11,7 @@
     <parent>
         <groupId>com.hedera</groupId>
         <artifactId>hedera-mirror-node</artifactId>
-        <version>0.15.0-rc1</version>
+        <version>0.15.0-rc2</version>
     </parent>
 
     <properties>

diff --git a/hedera-mirror-grpc/src/main/java/com/hedera/mirror/grpc/listener/NotifyingTopicListener.java b/hedera-mirror-grpc/src/main/java/com/hedera/mirror/grpc/listener/NotifyingTopicListener.java
@@ -27,10 +27,12 @@
 import io.vertx.pgclient.pubsub.PgChannel;
 import io.vertx.pgclient.pubsub.PgSubscriber;
 import java.time.Duration;
+import java.util.Objects;
 import javax.inject.Named;
 import lombok.extern.log4j.Log4j2;
 import reactor.core.publisher.EmitterProcessor;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.FluxSink;
 import reactor.core.scheduler.Schedulers;
 import reactor.util.retry.Retry;
 
@@ -60,7 +62,7 @@ public NotifyingTopicListener(DbProperties dbProperties, ListenerProperties list
         PgSubscriber subscriber = PgSubscriber.subscriber(vertx, connectOptions)
                 .reconnectPolicy(retries -> {
                     log.warn("Attempting reconnect");
-                    return frequency.toMillis();
+                    return frequency.toMillis() * Math.min(retries, 4);
                 });
 
         // Connect asynchronously to avoid crashing the application on startup if the database is down
@@ -76,9 +78,9 @@ public NotifyingTopicListener(DbProperties dbProperties, ListenerProperties list
         topicMessages = Flux.defer(() -> listen())
                 .publishOn(Schedulers.boundedElastic())
                 .map(this::toTopicMessage)
+                .filter(Objects::nonNull)
                 .name("notify")
                 .metrics()
-                .doFinally(s -> unlisten())
                 .doOnError(t -> log.error("Error listening for messages", t))
                 .retryWhen(Retry.backoff(Long.MAX_VALUE, frequency).maxBackoff(frequency.multipliedBy(4L)))
                 .share();
@@ -98,7 +100,8 @@ private boolean filterMessage(TopicMessage message, TopicMessageFilter filter) {
 
     private Flux<String> listen() {
         EmitterProcessor<String> emitterProcessor = EmitterProcessor.create();
-        channel.handler(json -> emitterProcessor.onNext(json));
+        FluxSink<String> sink = emitterProcessor.sink().onDispose(this::unlisten);
+        channel.handler(json -> sink.next(json));
         log.info("Listening for messages");
         return emitterProcessor;
     }
@@ -112,7 +115,9 @@ private TopicMessage toTopicMessage(String payload) {
         try {
             return objectMapper.readValue(payload, TopicMessage.class);
         } catch (Exception ex) {
-            throw new RuntimeException(ex);
+            // Discard invalid messages. No need to propagate error and cause a reconnect.
+            log.error("Error parsing message {}", payload, ex);
+            return null;
         }
     }
 }
diff --git a/hedera-mirror-grpc/src/main/resources/application.yml b/hedera-mirror-grpc/src/main/resources/application.yml
@@ -63,6 +63,7 @@ spring:
     properties:
       hibernate.criteria.literal_handling_mode: BIND # Ensure Criteria API queries use bind parameters and not literals
       hibernate.generate_statistics: true
+      hibernate.hbm2ddl.auto: none
   test:
     database:
       replace: NONE
diff --git a/hedera-mirror-importer/pom.xml b/hedera-mirror-importer/pom.xml
@@ -13,7 +13,7 @@
     <parent>
         <groupId>com.hedera</groupId>
         <artifactId>hedera-mirror-node</artifactId>
-        <version>0.15.0-rc1</version>
+        <version>0.15.0-rc2</version>
     </parent>
 
     <properties>

diff --git a/hedera-mirror-protobuf/pom.xml b/hedera-mirror-protobuf/pom.xml
@@ -11,7 +11,7 @@
     <parent>
         <groupId>com.hedera</groupId>
         <artifactId>hedera-mirror-node</artifactId>
-        <version>0.15.0-rc1</version>
+        <version>0.15.0-rc2</version>
     </parent>
 
     <properties>

diff --git a/hedera-mirror-rest/monitoring/monitor_apis/package-lock.json b/hedera-mirror-rest/monitoring/monitor_apis/package-lock.json
diff --git a/hedera-mirror-rest/monitoring/monitor_apis/package.json b/hedera-mirror-rest/monitoring/monitor_apis/package.json
@@ -1,6 +1,6 @@
 {
   "name": "hedera-mirror-monitor",
-  "version": "0.15.0-rc1",
+  "version": "0.15.0-rc2",
   "description": "Hedera Mirror Node Monitor",
   "main": "server.js",
   "private": true,

diff --git a/hedera-mirror-rest/package-lock.json b/hedera-mirror-rest/package-lock.json
diff --git a/hedera-mirror-rest/package.json b/hedera-mirror-rest/package.json
@@ -1,6 +1,6 @@
 {
   "name": "hedera-mirror-rest",
-  "version": "0.15.0-rc1",
+  "version": "0.15.0-rc2",
   "description": "Hedera Mirror Node REST API",
   "main": "server.js",
   "private": true,

diff --git a/hedera-mirror-rest/pom.xml b/hedera-mirror-rest/pom.xml
@@ -11,7 +11,7 @@
     <parent>
         <groupId>com.hedera</groupId>
         <artifactId>hedera-mirror-node</artifactId>
-        <version>0.15.0-rc1</version>
+        <version>0.15.0-rc2</version>
     </parent>
 
     <build>

diff --git a/hedera-mirror-test/pom.xml b/hedera-mirror-test/pom.xml
@@ -12,7 +12,7 @@
     <parent>
         <groupId>com.hedera</groupId>
         <artifactId>hedera-mirror-node</artifactId>
-        <version>0.15.0-rc1</version>
+        <version>0.15.0-rc2</version>
     </parent>
 
     <properties>

diff --git a/hedera-mirror-test/src/test/resources/k8s/hcs-perf-message-submit.yml b/hedera-mirror-test/src/test/resources/k8s/hcs-perf-message-submit.yml
@@ -41,7 +41,7 @@ spec:
                     app.kubernetes.io/name: test
       restartPolicy: Never
       containers:
-        - image: gcr.io/mirrornode/hedera-mirror-test:0.15.0-rc1
+        - image: gcr.io/mirrornode/hedera-mirror-test:0.15.0-rc2
           name: test
           env:
             - name: testProfile

diff --git a/hedera-mirror-test/src/test/resources/k8s/hcs-perf-publish-test.yml b/hedera-mirror-test/src/test/resources/k8s/hcs-perf-publish-test.yml
@@ -43,7 +43,7 @@ spec:
                     app.kubernetes.io/name: test
       restartPolicy: Never
       containers:
-        - image: gcr.io/mirrornode/hedera-mirror-test:0.15.0-rc1
+        - image: gcr.io/mirrornode/hedera-mirror-test:0.15.0-rc2
           name: test
           env:
             - name: testProfile

diff --git a/hedera-mirror-test/src/test/resources/k8s/hcs-perf-subscribe-test.yml b/hedera-mirror-test/src/test/resources/k8s/hcs-perf-subscribe-test.yml
@@ -46,7 +46,7 @@ spec:
                     app.kubernetes.io/name: test
       restartPolicy: Never
       containers:
-        - image: gcr.io/mirrornode/hedera-mirror-test:0.15.0-rc1
+        - image: gcr.io/mirrornode/hedera-mirror-test:0.15.0-rc2
           name: test
           env:
             - name: testProfile

diff --git a/pom.xml b/pom.xml
@@ -4,7 +4,7 @@
 
     <groupId>com.hedera</groupId>
     <artifactId>hedera-mirror-node</artifactId>
-    <version>0.15.0-rc1</version>
+    <version>0.15.0-rc2</version>
     <description>Hedera Mirror Node mirrors data from Hedera nodes and serves it via an API</description>
     <inceptionYear>2019</inceptionYear>
     <modelVersion>4.0.0</modelVersion>