Retry vitess connection refused errors 100 times

src/main/java/io/debezium/connector/vitess/VitessErrorHandler.java

@@ -12,8 +12,12 @@
 import io.debezium.pipeline.ErrorHandler;
 import io.grpc.StatusRuntimeException;
 
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
 public class VitessErrorHandler extends ErrorHandler {
     private static final Logger LOGGER = LoggerFactory.getLogger(VitessErrorHandler.class);
+    private static AtomicInteger connectionRefusedRestarts = new AtomicInteger(100);
 
     public VitessErrorHandler(VitessConnectorConfig connectorConfig, ChangeEventQueue<?> queue) {
         super(VitessConnector.class, connectorConfig, queue);
@@ -34,6 +38,9 @@ protected boolean isRetriable(Throwable throwable) {
                     }
                     return false;
                 case UNAVAILABLE:
+                    if (throwable.getCause().toString().contains("AnnotatedConnectException: Connection refused")) {
+                        return connectionRefusedRestarts.getAndDecrement() > 0;
+                    }
                     return true;
                 case UNKNOWN:
                     // Stream timeout error due to idle VStream or vstream ended unexpectedly.

src/test/java/io/debezium/connector/vitess/VitessErrorHandlerTest.java

@@ -0,0 +1,43 @@
+package io.debezium.connector.vitess;
+
+import io.debezium.connector.base.ChangeEventQueue;
+import io.debezium.pipeline.DataChangeEvent;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import org.junit.Test;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class VitessErrorHandlerTest  {
+
+    VitessConnectorConfig config = new VitessConnectorConfig(TestHelper.defaultConfig().build());
+    ChangeEventQueue<DataChangeEvent> queue = new ChangeEventQueue.Builder<DataChangeEvent>().build();
+    VitessErrorHandler vitessErrorHandler = new VitessErrorHandler(config, queue);
+
+    @Test
+    public void isRetriable_returnsTrueForConnectionRefusedExceptionThat() {
+        boolean result = vitessErrorHandler.isRetriable(createConnectionRefusedException());
+        assertTrue(result);
+    }
+
+    @Test
+    public void isRetriable_ReturnsFalseAfter100Retries() {
+        for (int i = 0; i < 100; i++) {
+            vitessErrorHandler.isRetriable(createConnectionRefusedException());
+        }
+        assertFalse(vitessErrorHandler.isRetriable(createConnectionRefusedException()));
+    }
+
+    private StatusRuntimeException createConnectionRefusedException() {
+        Throwable connectionRefused = new AnnotatedConnectException("Connection refused");
+        return Status.UNAVAILABLE.withDescription("io exception").withCause(connectionRefused).asRuntimeException();
+    }
+
+    // mocks com.ververica.cdc.connectors.vitess.shaded.io.netty.channel.AbstractChannel$AnnotatedConnectException
+    static class AnnotatedConnectException extends RuntimeException {
+        public AnnotatedConnectException(String message) {
+            super(message);
+        }
+    }
+}