feat: store the status of annotation signal in the PersistentVolume

Author: teoincontatto

stackgres-k8s/src/stream/src/main/java/io/stackgres/stream/jobs/source/DebeziumAnnotationSignalChannelReader.java

@@ -5,7 +5,6 @@
 
 package io.stackgres.stream.jobs.source;
 
-import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.file.Files;
@@ -17,6 +16,7 @@
 import java.util.Properties;
 import java.util.Set;
 
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import io.debezium.config.CommonConnectorConfig;
 import io.debezium.pipeline.signal.SignalRecord;
 import io.debezium.pipeline.signal.channels.SignalChannelReader;
@@ -27,9 +27,13 @@
 import io.stackgres.common.crd.sgstream.StackGresStream;
 import io.stackgres.common.resource.CustomResourceFinder;
 import io.stackgres.stream.app.StreamProperty;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class DebeziumAnnotationSignalChannelReader implements SignalChannelReader {
 
+  private static final Logger LOGGER = LoggerFactory.getLogger(DebeziumAnnotationSignalChannelReader.class);
+
   private static final String STACKGRES_IO_DEBEZIUM_SIGNAL_KEY_PREFIX =
       "debezium-signal." + StackGresContext.STACKGRES_KEY_PREFIX;
 
@@ -52,52 +56,53 @@ public String name() {
   public void init(CommonConnectorConfig connectorConfig) {
     this.streamName = StreamProperty.STREAM_NAME.getString();
     this.streamNamespace = StreamProperty.STREAM_NAMESPACE.getString();
-    this.annotationSignalPropertiesPath = Paths.get(StreamPath.DEBEZIUM_DATABASE_HISTORY_PATH.path());
+    this.annotationSignalPropertiesPath = Paths.get(StreamPath.DEBEZIUM_ANNOTATION_SIGNAL_PATH.path());
   }
 
+  @SuppressFBWarnings(value = "REC_CATCH_EXCEPTION",
+      justification = "Wanted behavior since executed directly from a thread")
   @Override
   public List<SignalRecord> read() {
-    Properties annotationSignalProperties = new Properties();
-    if (Files.exists(annotationSignalPropertiesPath)) {
-      try (InputStream inputStream = Files.newInputStream(annotationSignalPropertiesPath)) {
-        annotationSignalProperties.load(inputStream);
-      } catch (IOException ex) {
-        throw new RuntimeException(ex);
+    try {
+      Properties annotationSignalProperties = new Properties();
+      if (Files.exists(annotationSignalPropertiesPath)) {
+        try (InputStream inputStream = Files.newInputStream(annotationSignalPropertiesPath)) {
+          annotationSignalProperties.load(inputStream);
+        }
       }
-    }
-    final var streamSignalAnnotations =
-        streamFinder.findByNameAndNamespace(streamName, streamNamespace)
-        .map(HasMetadata::getMetadata)
-        .map(ObjectMeta::getAnnotations)
-        .stream()
-        .map(Map::entrySet)
-        .flatMap(Set::stream)
-        .filter(annotation -> annotation.getKey().startsWith(STACKGRES_IO_DEBEZIUM_SIGNAL_KEY_PREFIX))
-        .filter(annotation -> Optional
-            .ofNullable(annotationSignalProperties.getProperty(annotation.getKey()))
-            .map(value -> !annotation.getValue().equals(value))
-            .orElse(true));
-    streamSignalAnnotations.forEach(annotation -> annotationSignalProperties
-        .setProperty(annotation.getKey(), annotation.getValue()));
-    if (!Files.exists(annotationSignalPropertiesPath)) {
-      try {
+      final var streamSignalAnnotations =
+          streamFinder.findByNameAndNamespace(streamName, streamNamespace)
+          .map(HasMetadata::getMetadata)
+          .map(ObjectMeta::getAnnotations)
+          .stream()
+          .map(Map::entrySet)
+          .flatMap(Set::stream)
+          .filter(annotation -> annotation.getKey().startsWith(STACKGRES_IO_DEBEZIUM_SIGNAL_KEY_PREFIX))
+          .filter(annotation -> Optional
+              .ofNullable(annotationSignalProperties.getProperty(annotation.getKey()))
+              .map(value -> !annotation.getValue().equals(value))
+              .orElse(true))
+          .toList();
+      streamSignalAnnotations.forEach(annotation -> annotationSignalProperties
+          .setProperty(annotation.getKey(), annotation.getValue()));
+      if (!Files.exists(annotationSignalPropertiesPath)) {
         Files.createFile(annotationSignalPropertiesPath);
-      } catch (IOException ex) {
-        throw new RuntimeException(ex);
       }
+      try (OutputStream outputStream = Files.newOutputStream(annotationSignalPropertiesPath)) {
+        annotationSignalProperties.store(outputStream, null);
+      }
+      return streamSignalAnnotations
+          .stream()
+          .map(annotation -> new SignalRecord(
+              annotation.getKey(),
+              extractType(annotation.getKey()),
+              annotation.getValue(),
+              Map.of()))
+          .toList();
+    } catch (Exception ex) {
+      LOGGER.error("An error ocured while loading SGStream annotations: {}", ex.getMessage(), ex);
+      return List.of();
     }
-    try (OutputStream outputStream = Files.newOutputStream(annotationSignalPropertiesPath)) {
-      annotationSignalProperties.store(outputStream, null);
-    } catch (IOException ex) {
-      throw new RuntimeException(ex);
-    }
-    return streamSignalAnnotations
-        .map(annotation -> new SignalRecord(
-            annotation.getKey(),
-            extractType(annotation.getKey()),
-            annotation.getValue(),
-            Map.of()))
-        .toList();
   }
 
   private String extractType(String key) {