Merge release-1.2.2 to main (#640)

* Handle the underlying errors from the Horizon server connection (#635)

* Add silence watcher that sets the observer to SILENCE_ERROR when the SSEStream is silent for too long.
* Add SSEStream failure handler that sets the observer to STREAM_ERROR when failed.
* Add status watcher that decides to reconnect or to set the health status to RED depends on the reconnection/backoff timer status.

* Increased version to 1.2.2 (#639)

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## 1.2.2
+* Detects and handle silent and errored SSEStream. [#632](https://github.com/stellar/java-stellar-anchor-sdk/issues/632)
+* When the health status is RED, set the status code to 500.
+
 ## 1.2.1
 * Fix gson serialization error of the Refunds object. [#626](https://github.com/stellar/java-stellar-anchor-sdk/issues/626) 
 

anchor-reference-server/src/main/java/org/stellar/anchor/reference/event/KafkaListener.java

@@ -43,8 +43,8 @@ Consumer<String, AnchorEvent> createKafkaConsumer() {
 
     props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaListenerSettings.getBootStrapServer());
     props.put(ConsumerConfig.GROUP_ID_CONFIG, "group_one1");
-    // props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");  // start reading from
-    // earliest available message
+    // props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+    // start reading from the earliest available message
     props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
     props.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
     props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
@@ -141,7 +141,7 @@ public HealthCheckResult check() {
 
     return KafkaHealthCheckResult.builder()
         .name(getName())
-        .status(status.getName())
+        .status(status)
         .kafkaAvailable(kafkaAvailable)
         .running(!executor.isTerminated())
         .build();
@@ -162,9 +162,9 @@ boolean validateKafka() {
 class KafkaHealthCheckResult implements HealthCheckResult {
   transient String name;
 
-  List<String> statuses = List.of(GREEN.getName(), RED.getName());
+  List<HealthCheckStatus> statuses = List.of(GREEN, RED);
 
-  String status;
+  HealthCheckStatus status;
 
   boolean running;
 

api-schema/src/main/java/org/stellar/anchor/api/platform/HealthCheckResult.java

@@ -10,12 +10,12 @@ public interface HealthCheckResult {
    *
    * @return the list of health check status.
    */
-  List<String> getStatuses();
+  List<HealthCheckStatus> getStatuses();
 
   /**
    * the current status of the health check result.
    *
    * @return the status.
    */
-  String getStatus();
+  HealthCheckStatus getStatus();
 }

build.gradle.kts

@@ -118,7 +118,7 @@ subprojects {
 
 allprojects {
   group = "org.stellar.anchor-sdk"
-  version = "1.2.1"
+  version = "1.2.2"
 
   tasks.jar {
     manifest {

core/src/main/java/org/stellar/anchor/util/ExponentialBackoffTimer.java

@@ -42,4 +42,12 @@ public void reset() {
   public void sleep() throws InterruptedException {
     Thread.sleep(sleepSeconds * 1000);
   }
+
+  public long currentTimer() {
+    return sleepSeconds;
+  }
+
+  public boolean isTimerMaxed() {
+    return sleepSeconds >= maxSleepSeconds;
+  }
 }

docs/resources/docker-examples/kafka/docker-compose.yaml

@@ -1,4 +1,4 @@
-version: '2'
+version: '2.4'
 services:
   zookeeper:
     platform: linux/x86_64

helm-charts/sep-service/templates/ingress.yaml

@@ -46,56 +46,3 @@ spec:
                 number: {{ tpl .backend.servicePort $ }}
   {{- end }}
   {{- end }}
-
-# Stellar Observer
-{{- $root := . }}
-{{- $stellarObserver := (.Values.stellarObserver) }}
-{{- if $stellarObserver.enabled }}
----
-apiVersion: networking.k8s.io/v1
-kind: Ingress
-metadata:
-  name: {{ .Values.fullName }}-ing-{{ .Values.service.name }}-observer
-  {{- if .Values.ingress.metadata }}
-  {{- range $key, $value := .Values.ingress.metadata }}
-  {{- end }}
-  {{- end }}
-  {{- if .Values.ingress.annotations }}
-  annotations:
-    {{- range $key, $value := .Values.ingress.annotations }}
-    {{ $key }}: {{ $value | quote }}
-    {{- end }}
-  {{- end }}
-  labels:
-    app.kubernetes.io/name: {{ .Values.fullName }}-ing-{{ .Values.service.name }}-observer
-    helm.sh/chart: {{ $.Chart.Name }}-{{ $.Chart.Version }}
-    app.kubernetes.io/instance: {{ .Release.Name }}
-    app.kubernetes.io/managed-by: {{ .Release.Service }}  
-spec:
-  {{- if .Values.ingress.ingressClassName }}
-  ingressClassName: {{ .Values.ingress.ingressClassName }}
-  {{- end }}
-  {{- if .Values.ingress.tls }}
-  tls:
-  - hosts:
-    - {{ $stellarObserver.ingress.tls.host }}
-    {{- if .Values.ingress.tls.secretName }}
-    secretName: {{ .Values.ingress.tls.secretName }}
-    {{- end }}      
-  {{- end }}
-  {{- if .Values.ingress.rules }}
-  rules:
-  {{- range $stellarObserver.ingress.rules.hosts }}
-  - host: {{ tpl .host $ }}
-    http:
-      paths:
-        - path: {{ .path }}
-          pathType: {{ .pathType | quote }}
-          backend:
-            service:
-              name: {{ $root.Values.fullName }}-svc-{{ $root.Values.service.name }}-observer
-              port:
-                number: {{ $stellarObserver.deployment.port | default 8083 }}
-  {{- end }}
-  {{- end }}
-{{- end }}

integration-tests/src/test/kotlin/org/stellar/anchor/platform/AnchorPlatformIntegrationTest.kt

@@ -320,7 +320,6 @@ class AnchorPlatformIntegrationTest {
         "sep38.quoteIntegrationEndPoint" to "http://localhost:8081",
         "payment-gateway.circle.name" to "circle",
         "payment-gateway.circle.enabled" to "true",
-        "spring.jpa.database-platform" to "org.stellar.anchor.platform.sqlite.SQLiteDialect",
         "logging.level.root" to "INFO"
       )
 
@@ -393,16 +392,16 @@ class AnchorPlatformIntegrationTest {
 
     val stellarPaymentObserverCheck = checks["stellar_payment_observer"] as Map<*, *>
     assertEquals(2, stellarPaymentObserverCheck.size)
-    assertEquals("green", stellarPaymentObserverCheck["status"])
+    assertEquals("GREEN", stellarPaymentObserverCheck["status"])
 
     val observerStreams = stellarPaymentObserverCheck["streams"] as List<*>
     assertEquals(1, observerStreams.size)
 
     val stream1 = observerStreams[0] as Map<*, *>
-    assertEquals(4, stream1.size)
+    assertEquals(5, stream1.size)
     assertEquals(false, stream1["thread_shutdown"])
     assertEquals(false, stream1["thread_terminated"])
     assertEquals(false, stream1["stopped"])
-    assertNotNull(stream1["lastEventId"])
+    assertNotNull(stream1["last_event_id"])
   }
 }

integration-tests/src/test/resources/integration-test.anchor-config.yaml

@@ -8,7 +8,7 @@ stellar:
       settings: app-config                        # The location of the configuration data
     data-access:
       type: data-spring-jdbc                      # Activate [config-spring-jdbc] module.
-      settings: data-spring-jdbc-sqlite         # The location of the configuration data in this file.
+      settings: data-spring-jdbc-h2               # The location of the configuration data in this file.
     logging:
       type: logging-logback
       settings: logging-logback-settings

platform/src/main/java/org/stellar/anchor/platform/controller/HealthController.java

@@ -1,8 +1,11 @@
 package org.stellar.anchor.platform.controller;
 
 import java.util.List;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.*;
 import org.stellar.anchor.api.platform.HealthCheckResponse;
+import org.stellar.anchor.api.platform.HealthCheckStatus;
 import org.stellar.anchor.platform.service.HealthCheckService;
 
 @RestController
@@ -16,10 +19,20 @@ public class HealthController {
   }
 
   @RequestMapping(method = {RequestMethod.GET})
-  public HealthCheckResponse health(@RequestParam(required = false) List<String> checks) {
+  public ResponseEntity<HealthCheckResponse> health(
+      @RequestParam(required = false) List<String> checks) {
     if (checks == null) {
       checks = List.of("all");
     }
-    return healthCheckService.check(checks);
+    HealthCheckResponse healthCheckResponse = healthCheckService.check(checks);
+
+    boolean unhealthy =
+        healthCheckResponse.getChecks().values().stream()
+            .anyMatch(result -> result.getStatus() == HealthCheckStatus.RED);
+    if (unhealthy) {
+      return new ResponseEntity<>(healthCheckResponse, HttpStatus.INTERNAL_SERVER_ERROR);
+    } else {
+      return ResponseEntity.ok(healthCheckResponse);
+    }
   }
 }