Retry Policy for Jmx Connection #700

- Retry Policy for jmx connection
- Read all nodes statuses from nodes_sync table
- Filter out all node with unavailable status
- Retry to attempt jmx connection for unavailable nodes
- If connection successful add connection and change
status to available in table.
- If connection failed after given retry attempts, change
status to unreachable in table.
- This is schedule service that would execute after given
fix delay. By default it would run after every 24 hour and
scan the table.

application/src/main/java/com/ericsson/bss/cassandra/ecchronos/application/config/connection/ConnectionConfig.java

@@ -21,19 +21,19 @@ public class ConnectionConfig
     private DistributedNativeConnection myCqlConnection = new DistributedNativeConnection();
     private DistributedJmxConnection myJmxConnection = new DistributedJmxConnection();
 
-    @JsonProperty("cql")
+    @JsonProperty ("cql")
     public final DistributedNativeConnection getCqlConnection()
     {
         return myCqlConnection;
     }
 
-    @JsonProperty("jmx")
+    @JsonProperty ("jmx")
     public final DistributedJmxConnection getJmxConnection()
     {
         return myJmxConnection;
     }
 
-    @JsonProperty("cql")
+    @JsonProperty ("cql")
     public final void setCqlConnection(final DistributedNativeConnection cqlConnection)
     {
         if (cqlConnection != null)
@@ -42,7 +42,7 @@ public final void setCqlConnection(final DistributedNativeConnection cqlConnecti
         }
     }
 
-    @JsonProperty("jmx")
+    @JsonProperty ("jmx")
     public final void setJmxConnection(final DistributedJmxConnection jmxConnection)
     {
         if (jmxConnection != null)

application/src/main/java/com/ericsson/bss/cassandra/ecchronos/application/config/connection/DistributedJmxConnection.java

@@ -19,10 +19,14 @@
 
 import com.ericsson.bss.cassandra.ecchronos.connection.DistributedNativeConnectionProvider;
 import com.ericsson.bss.cassandra.ecchronos.data.sync.EccNodesSync;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
 import java.util.function.Supplier;
 
 public class DistributedJmxConnection extends Connection<DistributedJmxConnectionProvider>
 {
+    private RetryPolicyConfig myRetryPolicyConfig = new RetryPolicyConfig();
+
     public DistributedJmxConnection()
     {
         try
@@ -35,16 +39,28 @@ public DistributedJmxConnection()
         }
     }
 
+    @JsonProperty ("retryPolicy")
+    public final RetryPolicyConfig getRetryPolicyConfig()
+    {
+        return myRetryPolicyConfig;
+    }
+
+    @JsonProperty ("retryPolicy")
+    public final void setRetryPolicyConfig(final RetryPolicyConfig retryPolicyConfig)
+    {
+        myRetryPolicyConfig = retryPolicyConfig;
+    }
+
     /**
      * @return
      */
     @Override
     protected Class<?>[] expectedConstructor()
     {
         return new Class<?>[] {
-                Supplier.class,
-                DistributedNativeConnectionProvider.class,
-                EccNodesSync.class
+                                Supplier.class,
+                                DistributedNativeConnectionProvider.class,
+                                EccNodesSync.class
         };
     }
 }

application/src/main/java/com/ericsson/bss/cassandra/ecchronos/application/config/connection/RetryPolicyConfig.java

@@ -0,0 +1,157 @@
+/*
+ * Copyright 2024 Telefonaktiebolaget LM Ericsson
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.ericsson.bss.cassandra.ecchronos.application.config.connection;
+
+import java.util.Locale;
+import java.util.concurrent.TimeUnit;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public final class RetryPolicyConfig
+{
+
+    private static final int DEFAULT_MAX_ATTEMPTS = 5;
+    private static final long DEFAULT_DELAY = 5000;
+    private static final long DEFAULT_MAX_DELAY = 30000;
+    private static final long DEFAULT_INITIAL_DELAY = 86400;
+    private static final long DEFAULT_FIXED_DELAY = 86400;
+
+    @JsonProperty ("maxAttempts")
+    private Integer myMaxAttempts = DEFAULT_MAX_ATTEMPTS;
+    @JsonProperty ("delay")
+    private long myDelay = DEFAULT_DELAY;
+    @JsonProperty ("maxDelay")
+    private long myMaxDelay = DEFAULT_MAX_DELAY;
+    @JsonProperty ("unit")
+    private String myUnit = "seconds";
+    @JsonProperty ("initialDelay")
+    private long myInitialDelay = DEFAULT_INITIAL_DELAY;
+    @JsonProperty ("fixedDelay")
+    private long myFixedDelay = DEFAULT_FIXED_DELAY;
+
+    public RetryPolicyConfig()
+    {
+    }
+
+    public RetryPolicyConfig(final Integer maxAttempts,
+                             final Integer delay,
+                             final Integer maxDelay,
+                             final String unit,
+                             final long initialDelay,
+                             final long fixedDelay)
+    {
+        this.myMaxAttempts = maxAttempts;
+        this.myDelay = convertToMillis(delay, unit);
+        this.myMaxDelay = convertToMillis(maxDelay, unit);
+        this.myUnit = unit;
+        this.myInitialDelay = initialDelay;
+        this.myFixedDelay = fixedDelay;
+    }
+
+    @JsonProperty ("maxAttempts")
+    public Integer getMaxAttempts()
+    {
+        return myMaxAttempts;
+    }
+
+    @JsonProperty ("maxAttempts")
+    public void setMaxAttempts(final Integer maxAttempts)
+    {
+        this.myMaxAttempts = maxAttempts;
+    }
+
+    @JsonProperty ("delay")
+    public long getDelay()
+    {
+        return myDelay;
+    }
+
+    @JsonProperty ("delay")
+    public void setDelay(final Integer delay)
+    {
+        this.myDelay = convertToMillis(delay, myUnit);
+    }
+
+    @JsonProperty ("maxDelay")
+    public long getMaxDelay()
+    {
+        return myMaxDelay;
+    }
+
+    @JsonProperty ("maxDelay")
+    public void setMaxDelay(final Integer maxDelay)
+    {
+        this.myMaxDelay = convertToMillis(maxDelay, myUnit);
+    }
+
+    @JsonProperty ("unit")
+    public String getUnit()
+    {
+        return myUnit;
+    }
+
+    @JsonProperty ("unit")
+    public void setUnit(final String unit)
+    {
+        this.myUnit = unit;
+        // Recalculate delays with the new unit
+        this.myDelay = convertToMillis((int) TimeUnit.MILLISECONDS.toSeconds(this.myDelay), unit);
+        this.myMaxDelay = convertToMillis((int) TimeUnit.MILLISECONDS.toSeconds(this.myMaxDelay), unit);
+    }
+
+    @JsonProperty ("initialDelay")
+    public long getInitialDelay()
+    {
+        return myInitialDelay;
+    }
+
+    @JsonProperty ("initialDelay")
+    public void setInitialDelay(final Integer initialDelay)
+    {
+        this.myInitialDelay = convertToMillis(initialDelay, myUnit);
+    }
+
+    @JsonProperty ("fixedDelay")
+    public long getFixedDelay()
+    {
+        return myFixedDelay;
+    }
+
+    @JsonProperty ("fixedDelay")
+    public void setFixedDelay(final Integer fixedDelay)
+    {
+        this.myFixedDelay = convertToMillis(fixedDelay, myUnit);
+    }
+
+    private long convertToMillis(final Integer value, final String unit)
+    {
+        return switch (unit.toLowerCase(Locale.ENGLISH))
+        {
+            case "milliseconds" -> value;
+            case "seconds" -> TimeUnit.SECONDS.toMillis(value);
+            case "minutes" -> TimeUnit.MINUTES.toMillis(value);
+            default -> throw new IllegalArgumentException("Unsupported time unit: " + unit);
+        };
+    }
+
+    public long currentDelay(final Integer count)
+    {
+        long currentDelay = myDelay * count;
+        if (currentDelay > myMaxDelay)
+        {
+            currentDelay = myMaxDelay;
+        }
+        return currentDelay;
+    }
+}

application/src/main/java/com/ericsson/bss/cassandra/ecchronos/application/spring/BeanConfigurator.java

@@ -208,6 +208,15 @@ public DistributedJmxConnectionProvider distributedJmxConnectionProvider(
                 jmxSecurity::get, distributedNativeConnectionProvider, eccNodesSync);
     }
 
+    @Bean
+    public RetrySchedulerService retrySchedulerService(final Config config,
+                                                       final DistributedJmxConnectionProvider jmxConnectionProvider,
+                                                       final EccNodesSync eccNodesSync,
+                                                       final DistributedNativeConnectionProvider nativeConnectionProvider)
+    {
+        return new RetrySchedulerService(eccNodesSync, config, jmxConnectionProvider, nativeConnectionProvider);
+    }
+
     private Security getSecurityConfig() throws ConfigurationException
     {
         return ConfigurationHelper.DEFAULT_INSTANCE.getConfiguration(SECURITY_FILE, Security.class);

application/src/main/java/com/ericsson/bss/cassandra/ecchronos/application/spring/RetryBackoffStrategy.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2024 Telefonaktiebolaget LM Ericsson
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.ericsson.bss.cassandra.ecchronos.application.spring;
+
+import com.ericsson.bss.cassandra.ecchronos.application.config.connection.RetryPolicyConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public final class RetryBackoffStrategy
+{
+    private static final Logger LOG = LoggerFactory.getLogger(RetryBackoffStrategy.class);
+    private static final int MILLISECONDS_1000 = 1000;
+    private final RetryPolicyConfig myRetryPolicyConfig;
+
+    public RetryBackoffStrategy(final RetryPolicyConfig retryPolicyConfig)
+    {
+        this.myRetryPolicyConfig = retryPolicyConfig;
+    }
+
+    public long getInitialDelay()
+    {
+        return myRetryPolicyConfig.getInitialDelay();
+    }
+
+    public long getFixedDelay()
+    {
+        return myRetryPolicyConfig.getFixedDelay();
+    }
+
+    public int getMaxAttempts()
+    {
+        return myRetryPolicyConfig.getMaxAttempts();
+    }
+
+    public long calculateDelay(final int attempt)
+    {
+        long baseDelay = myRetryPolicyConfig.getDelay();
+        long calculatedDelay = baseDelay * (attempt * 2L);
+        LOG.debug("Calculated delay for attempt {}: {} ms", attempt, calculatedDelay);
+        return Math.min(calculatedDelay, myRetryPolicyConfig.getMaxDelay() * MILLISECONDS_1000);
+    }
+
+    public void sleepBeforeNextRetry(final long delayMillis)
+    {
+        try
+        {
+            Thread.sleep(delayMillis);
+        }
+        catch (InterruptedException e)
+        {
+            LOG.error("Interrupted during sleep between retry attempts", e);
+            Thread.currentThread().interrupt();
+        }
+    }
+}