Repository: hazelcast/hazelcast-kubernetes
Pull request #122 (2018-12-28 10:25:46)

Add ZONE_AWARE
Changes:
- Add retrieving zone name from Kubernetes API
- Refactor `HazelcastKubernetesDiscoveryStrategy` to include `namespace` and `KubernetesClient`
- Add non-retryable keywords to `RetryUtils` (prevents retrying exceptions caused by "reason:Forbidden")


Patch:
diff --git a/pom.xml b/pom.xml
index 50f86e6..faa5db4 100644
--- a/pom.xml
+++ b/pom.xml
@@ -198,7 +198,7 @@
               com.hazelcast.spi.discovery,com.hazelcast.core,
               com.hazelcast.logging,com.hazelcast.nio, com.hazelcast.util,
               com.hazelcast.internal.json,javax.net.ssl,javax.security.auth.x500,
-              javax.naming, javax.naming.directory
+              javax.naming, javax.naming.directory, com.hazelcast.spi.partitiongroup
             </Import-Package>
           </instructions>
         </configuration>
diff --git a/src/main/java/com/hazelcast/kubernetes/DefaultKubernetesClient.java b/src/main/java/com/hazelcast/kubernetes/DefaultKubernetesClient.java
index 22b8ed2..52783a1 100644
--- a/src/main/java/com/hazelcast/kubernetes/DefaultKubernetesClient.java
+++ b/src/main/java/com/hazelcast/kubernetes/DefaultKubernetesClient.java
@@ -91,6 +91,17 @@ class DefaultKubernetesClient
         return parseEndpoint(json);
     }
 
+    @Override
+    public String zone(String namespace, String podName) {
+        String podUrlString = String.format("%s/api/v1/namespaces/%s/pods/%s", kubernetesMaster, namespace, podName);
+        JsonObject podJson = callGet(podUrlString);
+        String nodeName = parseNodeName(podJson);
+
+        String nodeUrlString = String.format("%s/api/v1/nodes/%s", kubernetesMaster, nodeName);
+        JsonObject nodeJson = callGet(nodeUrlString);
+        return parseZone(nodeJson);
+    }
+
     private JsonObject callGet(String urlString) {
         HttpURLConnection connection = null;
         try {
@@ -230,6 +241,19 @@ class DefaultKubernetesClient
         return result;
     }
 
+    private static String parseNodeName(JsonObject json) {
+        return toString(json.get("spec").asObject().get("nodeName"));
+    }
+
+    private static String parseZone(JsonObject json) {
+        JsonObject labels = json.get("metadata").asObject().get("labels").asObject();
+        JsonValue zone = labels.get("failure-domain.kubernetes.io/zone");
+        if (zone != null) {
+            return toString(zone);
+        }
+        return toString(labels.get("failure-domain.beta.kubernetes.io/zone"));
+    }
+
     private static JsonArray toJsonArray(JsonValue jsonValue) {
         if (jsonValue == null || jsonValue.isNull()) {
             return new JsonArray();
diff --git a/src/main/java/com/hazelcast/kubernetes/HazelcastKubernetesDiscoveryStrategy.java b/src/main/java/com/hazelcast/kubernetes/HazelcastKubernetesDiscoveryStrategy.java
index 9b72eea..8b80407 100644
--- a/src/main/java/com/hazelcast/kubernetes/HazelcastKubernetesDiscoveryStrategy.java
+++ b/src/main/java/com/hazelcast/kubernetes/HazelcastKubernetesDiscoveryStrategy.java
@@ -18,12 +18,20 @@ package com.hazelcast.kubernetes;
 
 import com.hazelcast.config.properties.PropertyDefinition;
 import com.hazelcast.logging.ILogger;
+import com.hazelcast.nio.IOUtil;
 import com.hazelcast.spi.discovery.AbstractDiscoveryStrategy;
 import com.hazelcast.spi.discovery.DiscoveryNode;
+import com.hazelcast.spi.partitiongroup.PartitionGroupMetaData;
 import com.hazelcast.util.StringUtil;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -45,8 +53,13 @@ final class HazelcastKubernetesDiscoveryStrategy
     private static final String DEFAULT_MASTER_URL = "https://kubernetes.default.svc";
     private static final int DEFAULT_SERVICE_DNS_TIMEOUT_SECONDS = 5;
 
+    private final String namespace;
+
+    private final KubernetesClient client;
     private final EndpointResolver endpointResolver;
 
+    private final Map<String, Object> memberMetadata = new HashMap<String, Object>();
+
     HazelcastKubernetesDiscoveryStrategy(ILogger logger, Map<String, Comparable> properties) {
         super(logger, properties);
 
@@ -57,10 +70,13 @@ final class HazelcastKubernetesDiscoveryStrategy
         int port = getOrDefault(properties, KUBERNETES_SYSTEM_PREFIX, SERVICE_PORT, 0);
         String serviceLabel = getOrNull(properties, KUBERNETES_SYSTEM_PREFIX, SERVICE_LABEL_NAME);
         String serviceLabelValue = getOrDefault(properties, KUBERNETES_SYSTEM_PREFIX, SERVICE_LABEL_VALUE, "true");
-        String namespace = getOrDefault(properties, KUBERNETES_SYSTEM_PREFIX, NAMESPACE, getNamespaceOrDefault());
+        namespace = getOrDefault(properties, KUBERNETES_SYSTEM_PREFIX, NAMESPACE, getNamespaceOrDefault());
         Boolean resolveNotReadyAddresses = getOrDefault(properties, KUBERNETES_SYSTEM_PREFIX, RESOLVE_NOT_READY_ADDRESSES, false);
         String kubernetesMaster = getOrDefault(properties, KUBERNETES_SYSTEM_PREFIX, KUBERNETES_MASTER_URL, DEFAULT_MASTER_URL);
         String apiToken = getOrDefault(properties, KUBERNETES_SYSTEM_PREFIX, KUBERNETES_API_TOKEN, null);
+        if (apiToken == null) {
+            apiToken = readAccountToken();
+        }
 
         logger.info("Kubernetes Discovery properties: { "
                 + "service-dns: " + serviceDns + ", "
@@ -73,12 +89,13 @@ final class HazelcastKubernetesDiscoveryStrategy
                 + "resolve-not-ready-addresses: " + resolveNotReadyAddresses + ", "
                 + "kubernetes-master: " + kubernetesMaster + "}");
 
+        client = buildKubernetesClient(apiToken, kubernetesMaster);
         EndpointResolver endpointResolver;
         if (serviceDns != null) {
             endpointResolver = new DnsEndpointResolver(logger, serviceDns, port, serviceDnsTimeout);
         } else {
             endpointResolver = new KubernetesApiEndpointResolver(logger, serviceName, port, serviceLabel, serviceLabelValue,
-                    namespace, resolveNotReadyAddresses, kubernetesMaster, apiToken);
+                    namespace, resolveNotReadyAddresses, client);
         }
         logger.info("Kubernetes Discovery activated resolver: " + endpointResolver.getClass().getSimpleName());
         this.endpointResolver = endpointResolver;
@@ -99,6 +116,36 @@ final class HazelcastKubernetesDiscoveryStrategy
         endpointResolver.start();
     }
 
+    @Override
+    public Map<String, Object> discoverLocalMetadata() {
+        if (memberMetadata.isEmpty()) {
+            memberMetadata.put(PartitionGroupMetaData.PARTITION_GROUP_ZONE, discoverZone());
+        }
+        return memberMetadata;
+    }
+
+    private String discoverZone() {
+        try {
+            String podName = System.getenv("POD_NAME");
+            if (podName == null) {
+                podName = System.getenv("HOSTNAME");
+            }
+            if (podName == null) {
+                podName = InetAddress.getLocalHost().getHostName();
+            }
+            String zone = client.zone(namespace, podName);
+            if (zone != null) {
+                return zone;
+            }
+        } catch (Exception e) {
+            // only log the exception and the message, Hazelcast should still start
+            getLogger().finest(e);
+        }
+        getLogger().warning("Cannot fetch the current zone, ZONE_AWARE feature is disabled");
+        return "unknown";
+    }
+
+    @Override
     public Iterable<DiscoveryNode> discoverNodes() {
         return endpointResolver.resolve();
     }
@@ -188,4 +235,31 @@ final class HazelcastKubernetesDiscoveryStrategy
             return null;
         }
     }
+
+    private KubernetesClient buildKubernetesClient(String token, String kubernetesMaster) {
+        if (StringUtil.isNullOrEmpty(token)) {
+            token = readAccountToken();
+        }
+        return new RetryKubernetesClient(new DefaultKubernetesClient(kubernetesMaster, token));
+    }
+
+    @SuppressFBWarnings("DMI_HARDCODED_ABSOLUTE_FILENAME")
+    private static String readAccountToken() {
+        return readFileContents("/var/run/secrets/kubernetes.io/serviceaccount/token");
+    }
+
+    protected static String readFileContents(String tokenFile) {
+        InputStream is = null;
+        try {
+            File file = new File(tokenFile);
+            byte[] data = new byte[(int) file.length()];
+            is = new FileInputStream(file);
+            is.read(data);
+            return new String(data, "UTF-8");
+        } catch (IOException e) {
+            throw new RuntimeException("Could not get token file", e);
+        } finally {
+            IOUtil.closeResource(is);
+        }
+    }
 }
diff --git a/src/main/java/com/hazelcast/kubernetes/KubernetesApiEndpointResolver.java b/src/main/java/com/hazelcast/kubernetes/KubernetesApiEndpointResolver.java
index 51b43c1..df37789 100644
--- a/src/main/java/com/hazelcast/kubernetes/KubernetesApiEndpointResolver.java
+++ b/src/main/java/com/hazelcast/kubernetes/KubernetesApiEndpointResolver.java
@@ -21,16 +21,9 @@ import com.hazelcast.kubernetes.KubernetesClient.Endpoints;
 import com.hazelcast.kubernetes.KubernetesClient.EntrypointAddress;
 import com.hazelcast.logging.ILogger;
 import com.hazelcast.nio.Address;
-import com.hazelcast.nio.IOUtil;
 import com.hazelcast.spi.discovery.DiscoveryNode;
 import com.hazelcast.spi.discovery.SimpleDiscoveryNode;
-import com.hazelcast.util.StringUtil;
-import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.io.InputStream;
 import java.net.InetAddress;
 import java.util.ArrayList;
 import java.util.List;
@@ -47,7 +40,7 @@ class KubernetesApiEndpointResolver
     private final KubernetesClient client;
 
     KubernetesApiEndpointResolver(ILogger logger, String serviceName, int port, String serviceLabel, String serviceLabelValue,
-                                  String namespace, Boolean resolveNotReadyAddresses, String kubernetesMaster, String apiToken) {
+                                  String namespace, Boolean resolveNotReadyAddresses, KubernetesClient client) {
 
         super(logger);
 
@@ -57,14 +50,7 @@ class KubernetesApiEndpointResolver
         this.serviceLabel = serviceLabel;
         this.serviceLabelValue = serviceLabelValue;
         this.resolveNotReadyAddresses = resolveNotReadyAddresses;
-        this.client = buildKubernetesClient(apiToken, kubernetesMaster);
-    }
-
-    private KubernetesClient buildKubernetesClient(String token, String kubernetesMaster) {
-        if (StringUtil.isNullOrEmpty(token)) {
-            token = getAccountToken();
-        }
-        return new RetryKubernetesClient(new DefaultKubernetesClient(kubernetesMaster, token));
+        this.client = client;
     }
 
     @Override
@@ -116,24 +102,4 @@ class KubernetesApiEndpointResolver
         }
         return NetworkConfig.DEFAULT_PORT;
     }
-
-    @SuppressFBWarnings("DMI_HARDCODED_ABSOLUTE_FILENAME")
-    private String getAccountToken() {
-        return readFileContents("/var/run/secrets/kubernetes.io/serviceaccount/token");
-    }
-
-    protected static String readFileContents(String tokenFile) {
-        InputStream is = null;
-        try {
-            File file = new File(tokenFile);
-            byte[] data = new byte[(int) file.length()];
-            is = new FileInputStream(file);
-            is.read(data);
-            return new String(data, "UTF-8");
-        } catch (IOException e) {
-            throw new RuntimeException("Could not get token file", e);
-        } finally {
-            IOUtil.closeResource(is);
-        }
-    }
 }
diff --git a/src/main/java/com/hazelcast/kubernetes/KubernetesClient.java b/src/main/java/com/hazelcast/kubernetes/KubernetesClient.java
index 4015e3e..1f48efa 100644
--- a/src/main/java/com/hazelcast/kubernetes/KubernetesClient.java
+++ b/src/main/java/com/hazelcast/kubernetes/KubernetesClient.java
@@ -55,6 +55,19 @@ interface KubernetesClient {
      */
     Endpoints endpointsByName(String namespace, String endpointName);
 
+    /**
+     * Retrieves zone name for the given {@code namespace} and the given {@code podName}.
+     * <p>
+     * Note that the Kubernetes environment must provide such information as defined
+     * <a href="https://kubernetes.io/docs/reference/kubernetes-api/labels-annotations-taints">here</a>.
+     *
+     * @param namespace namespace name
+     * @param podName   POD name
+     * @return zone name
+     * @see <a href="https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.11">Kubernetes Endpoint API</a>
+     */
+    String zone(String namespace, String podName);
+
     /**
      * Result which stores the information about all addresses.
      */
diff --git a/src/main/java/com/hazelcast/kubernetes/RetryKubernetesClient.java b/src/main/java/com/hazelcast/kubernetes/RetryKubernetesClient.java
index 1b8251d..3d9aa22 100644
--- a/src/main/java/com/hazelcast/kubernetes/RetryKubernetesClient.java
+++ b/src/main/java/com/hazelcast/kubernetes/RetryKubernetesClient.java
@@ -16,6 +16,8 @@
 
 package com.hazelcast.kubernetes;
 
+import java.util.Arrays;
+import java.util.List;
 import java.util.concurrent.Callable;
 
 /**
@@ -24,6 +26,7 @@ import java.util.concurrent.Callable;
 class RetryKubernetesClient
         implements KubernetesClient {
     private static final int DEFAULT_RETRIES = 10;
+    private static final List<String> NON_RETRYABLE_KEYWORDS = Arrays.asList("\"reason\":\"Forbidden\"");
 
     private final KubernetesClient kubernetesClient;
     private final int retries;
@@ -45,7 +48,7 @@ class RetryKubernetesClient
                     throws Exception {
                 return kubernetesClient.endpoints(namespace);
             }
-        }, retries);
+        }, retries, NON_RETRYABLE_KEYWORDS);
     }
 
     @Override
@@ -56,7 +59,7 @@ class RetryKubernetesClient
                     throws Exception {
                 return kubernetesClient.endpointsByLabel(namespace, serviceLabel, serviceLabelValue);
             }
-        }, retries);
+        }, retries, NON_RETRYABLE_KEYWORDS);
     }
 
     @Override
@@ -67,6 +70,17 @@ class RetryKubernetesClient
                     throws Exception {
                 return kubernetesClient.endpointsByName(namespace, endpointName);
             }
-        }, retries);
+        }, retries, NON_RETRYABLE_KEYWORDS);
+    }
+
+    @Override
+    public String zone(final String namespace, final String podName) {
+        return RetryUtils.retry(new Callable<String>() {
+            @Override
+            public String call()
+                    throws Exception {
+                return kubernetesClient.zone(namespace, podName);
+            }
+        }, retries, NON_RETRYABLE_KEYWORDS);
     }
 }
diff --git a/src/main/java/com/hazelcast/kubernetes/RetryUtils.java b/src/main/java/com/hazelcast/kubernetes/RetryUtils.java
index b93a81d..7173aa8 100644
--- a/src/main/java/com/hazelcast/kubernetes/RetryUtils.java
+++ b/src/main/java/com/hazelcast/kubernetes/RetryUtils.java
@@ -21,6 +21,7 @@ import com.hazelcast.logging.ILogger;
 import com.hazelcast.logging.Logger;
 import com.hazelcast.util.ExceptionUtil;
 
+import java.util.List;
 import java.util.concurrent.Callable;
 
 /**
@@ -45,14 +46,14 @@ final class RetryUtils {
      * <p>
      * If {@code callable} throws an unchecked exception, it is wrapped into {@link HazelcastException}.
      */
-    public static <T> T retry(Callable<T> callable, int retries) {
+    public static <T> T retry(Callable<T> callable, int retries, List<String> nonRetryableKeywords) {
         int retryCount = 0;
         while (true) {
             try {
                 return callable.call();
             } catch (Exception e) {
                 retryCount++;
-                if (retryCount > retries) {
+                if (retryCount > retries || containsAnyOf(e, nonRetryableKeywords)) {
                     throw ExceptionUtil.rethrow(e);
                 }
                 long waitIntervalMs = backoffIntervalForRetry(retryCount);
@@ -64,6 +65,20 @@ final class RetryUtils {
         }
     }
 
+    private static boolean containsAnyOf(Exception e, List<String> nonRetryableKeywords) {
+        Throwable currentException = e;
+        while (currentException != null) {
+            String exceptionMessage = currentException.getMessage();
+            for (String keyword : nonRetryableKeywords) {
+                if (exceptionMessage != null && exceptionMessage.contains(keyword)) {
+                    return true;
+                }
+            }
+            currentException = currentException.getCause();
+        }
+        return false;
+    }
+
     private static long backoffIntervalForRetry(int retryCount) {
         long result = INITIAL_BACKOFF_MS;
         for (int i = 1; i < retryCount; i++) {


Test patch:
diff --git a/src/test/java/com/hazelcast/kubernetes/DefaultKubernetesClientTest.java b/src/test/java/com/hazelcast/kubernetes/DefaultKubernetesClientTest.java
index b0c7fb2..371b041 100644
--- a/src/test/java/com/hazelcast/kubernetes/DefaultKubernetesClientTest.java
+++ b/src/test/java/com/hazelcast/kubernetes/DefaultKubernetesClientTest.java
@@ -35,6 +35,7 @@ import static com.github.tomakehurst.wiremock.client.WireMock.urlPathMatching;
 import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
 public class DefaultKubernetesClientTest {
@@ -51,6 +52,7 @@ public class DefaultKubernetesClientTest {
     private static final String SAMPLE_IP_PORT_1 = ipPort(SAMPLE_ADDRESS_1, SAMPLE_PORT_1);
     private static final String SAMPLE_IP_PORT_2 = ipPort(SAMPLE_ADDRESS_2, SAMPLE_PORT_2);
     private static final String SAMPLE_NOT_READY_IP = ipPort(SAMPLE_NOT_READY_ADDRESS, null);
+    private static final String ZONE = "us-central1-a";
 
     @Rule
     public WireMockRule wireMockRule = new WireMockRule(wireMockConfig().dynamicPort());
@@ -113,6 +115,44 @@ public class DefaultKubernetesClientTest {
         assertTrue(result.getNotReadyAddresses().isEmpty());
     }
 
+    @Test
+    public void zoneBeta() {
+        // given
+        String podName = "my-release-hazelcast-0";
+        String nodeName = "gke-rafal-test-cluster-default-pool-9238654c-12tz";
+        stubFor(get(urlPathMatching(String.format("/api/v1/namespaces/%s/pods/%s", NAMESPACE, podName)))
+                .withHeader("Authorization", equalTo(String.format("Bearer %s", TOKEN)))
+                .willReturn(aResponse().withStatus(200).withBody(podBody(nodeName))));
+        stubFor(get(urlPathMatching(String.format("/api/v1/nodes/%s", nodeName)))
+                .withHeader("Authorization", equalTo(String.format("Bearer %s", TOKEN)))
+                .willReturn(aResponse().withStatus(200).withBody(nodeBetaBody())));
+
+        // when
+        String zone = kubernetesClient.zone(NAMESPACE, podName);
+
+        // then
+        assertEquals(ZONE, zone);
+    }
+
+    @Test
+    public void zone() {
+        // given
+        String podName = "my-release-hazelcast-0";
+        String nodeName = "gke-rafal-test-cluster-default-pool-9238654c-12tz";
+        stubFor(get(urlPathMatching(String.format("/api/v1/namespaces/%s/pods/%s", NAMESPACE, podName)))
+                .withHeader("Authorization", equalTo(String.format("Bearer %s", TOKEN)))
+                .willReturn(aResponse().withStatus(200).withBody(podBody(nodeName))));
+        stubFor(get(urlPathMatching(String.format("/api/v1/nodes/%s", nodeName)))
+                .withHeader("Authorization", equalTo(String.format("Bearer %s", TOKEN)))
+                .willReturn(aResponse().withStatus(200).withBody(nodeBody())));
+
+        // when
+        String zone = kubernetesClient.zone(NAMESPACE, podName);
+
+        // then
+        assertEquals(ZONE, zone);
+    }
+
     @Test(expected = KubernetesClientException.class)
     public void forbidden() {
         // given
@@ -751,6 +791,205 @@ public class DefaultKubernetesClientTest {
                 , SAMPLE_ADDRESS_1, SAMPLE_PORT_1, SAMPLE_ADDRESS_2, SAMPLE_PORT_2);
     }
 
+    /**
+     * Real response recorded from the Kubernetes API call "/api/v1/namespaces/{namespace}/pods/{pod-name}".
+     */
+    private static String podBody(String nodeName) {
+        return String.format(
+                "{  \"kind\": \"Pod\",\n"
+                        + "  \"apiVersion\": \"v1\",\n"
+                        + "  \"metadata\": {\n"
+                        + "    \"name\": \"my-release-hazelcast-0\",\n"
+                        + "    \"generateName\": \"my-release-hazelcast-\",\n"
+                        + "    \"namespace\": \"default\",\n"
+                        + "    \"selfLink\": \"/api/v1/namespaces/default/pods/my-release-hazelcast-0\",\n"
+                        + "    \"uid\": \"53112ead-0511-11e9-9c53-42010a800013\",\n"
+                        + "    \"resourceVersion\": \"7724\",\n"
+                        + "    \"creationTimestamp\": \"2018-12-21T11:12:37Z\",\n"
+                        + "    \"labels\": {\n"
+                        + "      \"app\": \"hazelcast\",\n"
+                        + "      \"controller-revision-hash\": \"my-release-hazelcast-695d9d97dd\",\n"
+                        + "      \"release\": \"my-release\",\n"
+                        + "      \"role\": \"hazelcast\",\n"
+                        + "      \"statefulset.kubernetes.io/pod-name\": \"my-release-hazelcast-0\"\n"
+                        + "    },\n"
+                        + "    \"annotations\": {\n"
+                        + "      \"kubernetes.io/limit-ranger\": \"LimitRanger plugin set: cpu request for container my-release-hazelcast\"\n"
+                        + "    },\n"
+                        + "    \"ownerReferences\": [\n"
+                        + "      {\n"
+                        + "        \"apiVersion\": \"apps/v1\",\n"
+                        + "        \"kind\": \"StatefulSet\",\n"
+                        + "        \"name\": \"my-release-hazelcast\",\n"
+                        + "        \"uid\": \"53096f3b-0511-11e9-9c53-42010a800013\",\n"
+                        + "        \"controller\": true,\n"
+                        + "        \"blockOwnerDeletion\": true\n"
+                        + "      }\n"
+                        + "    ]\n"
+                        + "  },\n"
+                        + "  \"spec\": {\n"
+                        + "    \"volumes\": [\n"
+                        + "      {\n"
+                        + "        \"name\": \"hazelcast-storage\",\n"
+                        + "        \"configMap\": {\n"
+                        + "          \"name\": \"my-release-hazelcast-configuration\",\n"
+                        + "          \"defaultMode\": 420\n"
+                        + "        }\n"
+                        + "      },\n"
+                        + "      {\n"
+                        + "        \"name\": \"my-release-hazelcast-token-j89v4\",\n"
+                        + "        \"secret\": {\n"
+                        + "          \"secretName\": \"my-release-hazelcast-token-j89v4\",\n"
+                        + "          \"defaultMode\": 420\n"
+                        + "        }\n"
+                        + "      }\n"
+                        + "    ],\n"
+                        + "    \"containers\": [\n"
+                        + "      {\n"
+                        + "        \"name\": \"my-release-hazelcast\",\n"
+                        + "        \"image\": \"hazelcast/hazelcast:latest\",\n"
+                        + "        \"ports\": [\n"
+                        + "          {\n"
+                        + "            \"name\": \"hazelcast\",\n"
+                        + "            \"containerPort\": 5701,\n"
+                        + "            \"protocol\": \"TCP\"\n"
+                        + "          }\n"
+                        + "        ],\n"
+                        + "        \"env\": [\n"
+                        + "          {\n"
+                        + "            \"name\": \"JAVA_OPTS\",\n"
+                        + "            \"value\": \"-Dhazelcast.rest.enabled=true -Dhazelcast.config=/data/hazelcast/hazelcast.xml -DserviceName=my-release-hazelcast -Dnamespace=default -Dhazelcast.mancenter.enabled=true -Dhazelcast.mancenter.url=http://my-release-hazelcast-mancenter:8080/hazelcast-mancenter -Dhazelcast.shutdownhook.policy=GRACEFUL -Dhazelcast.shutdownhook.enabled=true -Dhazelcast.graceful.shutdown.max.wait=600 \"\n"
+                        + "          }\n"
+                        + "        ],\n"
+                        + "        \"resources\": {\n"
+                        + "          \"requests\": {\n"
+                        + "            \"cpu\": \"100m\"\n"
+                        + "          }\n"
+                        + "        },\n"
+                        + "        \"volumeMounts\": [\n"
+                        + "          {\n"
+                        + "            \"name\": \"hazelcast-storage\",\n"
+                        + "            \"mountPath\": \"/data/hazelcast\"\n"
+                        + "          },\n"
+                        + "          {\n"
+                        + "            \"name\": \"my-release-hazelcast-token-j89v4\",\n"
+                        + "            \"readOnly\": true,\n"
+                        + "            \"mountPath\": \"/var/run/secrets/kubernetes.io/serviceaccount\"\n"
+                        + "          }\n"
+                        + "        ],\n"
+                        + "        \"terminationMessagePath\": \"/dev/termination-log\",\n"
+                        + "        \"terminationMessagePolicy\": \"File\",\n"
+                        + "        \"imagePullPolicy\": \"Always\"\n"
+                        + "      }\n"
+                        + "    ],\n"
+                        + "    \"restartPolicy\": \"Always\",\n"
+                        + "    \"terminationGracePeriodSeconds\": 600,\n"
+                        + "    \"dnsPolicy\": \"ClusterFirst\",\n"
+                        + "    \"serviceAccountName\": \"my-release-hazelcast\",\n"
+                        + "    \"serviceAccount\": \"my-release-hazelcast\",\n"
+                        + "    \"nodeName\": \"%s\",\n"
+                        + "    \"securityContext\": {\n"
+                        + "      \"runAsUser\": 1001,\n"
+                        + "      \"fsGroup\": 1001\n"
+                        + "    },\n"
+                        + "    \"hostname\": \"my-release-hazelcast-0\",\n"
+                        + "    \"schedulerName\": \"default-scheduler\",\n"
+                        + "    \"tolerations\": [\n"
+                        + "      {\n"
+                        + "        \"key\": \"node.kubernetes.io/not-ready\",\n"
+                        + "        \"operator\": \"Exists\",\n"
+                        + "        \"effect\": \"NoExecute\",\n"
+                        + "        \"tolerationSeconds\": 300\n"
+                        + "      },\n"
+                        + "      {\n"
+                        + "        \"key\": \"node.kubernetes.io/unreachable\",\n"
+                        + "        \"operator\": \"Exists\",\n"
+                        + "        \"effect\": \"NoExecute\",\n"
+                        + "        \"tolerationSeconds\": 300\n"
+                        + "      }\n"
+                        + "    ]\n"
+                        + "  },\n"
+                        + "  \"status\": {\n"
+                        + "  }\n"
+                        + "}", nodeName);
+    }
+
+    /**
+     * Real response recorded from the Kubernetes (version < 1.13) API call "/api/v1/nodes/{node-name}".
+     */
+    private static String nodeBetaBody() {
+        return String.format(
+                "{"
+                        + "  \"kind\": \"Node\",\n"
+                        + "  \"apiVersion\": \"v1\",\n"
+                        + "  \"metadata\": {\n"
+                        + "    \"name\": \"gke-rafal-test-cluster-default-pool-9238654c-12tz\",\n"
+                        + "    \"selfLink\": \"/api/v1/nodes/gke-rafal-test-cluster-default-pool-9238654c-12tz\",\n"
+                        + "    \"uid\": \"ceab9c17-0508-11e9-9c53-42010a800013\",\n"
+                        + "    \"resourceVersion\": \"7954\",\n"
+                        + "    \"creationTimestamp\": \"2018-12-21T10:11:39Z\",\n"
+                        + "    \"labels\": {\n"
+                        + "      \"beta.kubernetes.io/arch\": \"amd64\",\n"
+                        + "      \"beta.kubernetes.io/fluentd-ds-ready\": \"true\",\n"
+                        + "      \"beta.kubernetes.io/instance-type\": \"n1-standard-1\",\n"
+                        + "      \"beta.kubernetes.io/os\": \"linux\",\n"
+                        + "      \"cloud.google.com/gke-nodepool\": \"default-pool\",\n"
+                        + "      \"cloud.google.com/gke-os-distribution\": \"cos\",\n"
+                        + "      \"failure-domain.beta.kubernetes.io/region\": \"us-central1\",\n"
+                        + "      \"failure-domain.beta.kubernetes.io/zone\": \"%s\",\n"
+                        + "      \"kubernetes.io/hostname\": \"gke-rafal-test-cluster-default-pool-9238654c-12tz\"\n"
+                        + "    },\n"
+                        + "    \"annotations\": {\n"
+                        + "      \"node.alpha.kubernetes.io/ttl\": \"0\",\n"
+                        + "      \"volumes.kubernetes.io/controller-managed-attach-detach\": \"true\"\n"
+                        + "    }\n"
+                        + "  },\n"
+                        + "  \"spec\": {\n"
+                        + "  },\n"
+                        + "  \"status\": {\n"
+                        + "  }\n"
+                        + "}", ZONE);
+    }
+
+    /**
+     * Real response recorded from the Kubernetes (version >= 1.13) API call "/api/v1/nodes/{node-name}".
+     */
+    private static String nodeBody() {
+        return String.format(
+                "{"
+                        + "  \"kind\": \"Node\",\n"
+                        + "  \"apiVersion\": \"v1\",\n"
+                        + "  \"metadata\": {\n"
+                        + "    \"name\": \"gke-rafal-test-cluster-default-pool-9238654c-12tz\",\n"
+                        + "    \"selfLink\": \"/api/v1/nodes/gke-rafal-test-cluster-default-pool-9238654c-12tz\",\n"
+                        + "    \"uid\": \"ceab9c17-0508-11e9-9c53-42010a800013\",\n"
+                        + "    \"resourceVersion\": \"7954\",\n"
+                        + "    \"creationTimestamp\": \"2018-12-21T10:11:39Z\",\n"
+                        + "    \"labels\": {\n"
+                        + "      \"beta.kubernetes.io/arch\": \"amd64\",\n"
+                        + "      \"beta.kubernetes.io/fluentd-ds-ready\": \"true\",\n"
+                        + "      \"beta.kubernetes.io/instance-type\": \"n1-standard-1\",\n"
+                        + "      \"beta.kubernetes.io/os\": \"linux\",\n"
+                        + "      \"cloud.google.com/gke-nodepool\": \"default-pool\",\n"
+                        + "      \"cloud.google.com/gke-os-distribution\": \"cos\",\n"
+                        + "      \"failure-domain.beta.kubernetes.io/region\": \"deprecated-region\",\n"
+                        + "      \"failure-domain.beta.kubernetes.io/zone\": \"deprecated-zone\",\n"
+                        + "      \"failure-domain.kubernetes.io/region\": \"us-central1\",\n"
+                        + "      \"failure-domain.kubernetes.io/zone\": \"%s\",\n"
+                        + "      \"kubernetes.io/hostname\": \"gke-rafal-test-cluster-default-pool-9238654c-12tz\"\n"
+                        + "    },\n"
+                        + "    \"annotations\": {\n"
+                        + "      \"node.alpha.kubernetes.io/ttl\": \"0\",\n"
+                        + "      \"volumes.kubernetes.io/controller-managed-attach-detach\": \"true\"\n"
+                        + "    }\n"
+                        + "  },\n"
+                        + "  \"spec\": {\n"
+                        + "  },\n"
+                        + "  \"status\": {\n"
+                        + "  }\n"
+                        + "}", ZONE);
+    }
+
     /**
      * Real response recorded from the Kubernetes API.
      */
diff --git a/src/test/java/com/hazelcast/kubernetes/KubernetesDiscoveryStrategyFactoryTest.java b/src/test/java/com/hazelcast/kubernetes/HazelcastKubernetesDiscoveryStrategyFactoryTest.java
similarity index 98%
rename from src/test/java/com/hazelcast/kubernetes/KubernetesDiscoveryStrategyFactoryTest.java
rename to src/test/java/com/hazelcast/kubernetes/HazelcastKubernetesDiscoveryStrategyFactoryTest.java
index c147185..cc560ea 100644
--- a/src/test/java/com/hazelcast/kubernetes/KubernetesDiscoveryStrategyFactoryTest.java
+++ b/src/test/java/com/hazelcast/kubernetes/HazelcastKubernetesDiscoveryStrategyFactoryTest.java
@@ -37,7 +37,7 @@ import static org.junit.Assert.assertTrue;
 
 @RunWith(PowerMockRunner.class)
 @PrepareForTest(KubernetesApiEndpointResolver.class)
-public class KubernetesDiscoveryStrategyFactoryTest {
+public class HazelcastKubernetesDiscoveryStrategyFactoryTest {
 
     private static final ILogger LOGGER = new NoLogFactory().getLogger("no");
     private static final String API_TOKEN = "token";
diff --git a/src/test/java/com/hazelcast/kubernetes/HazelcastKubernetesDiscoveryStrategyTest.java b/src/test/java/com/hazelcast/kubernetes/HazelcastKubernetesDiscoveryStrategyTest.java
new file mode 100644
index 0000000..bd8e656
--- /dev/null
+++ b/src/test/java/com/hazelcast/kubernetes/HazelcastKubernetesDiscoveryStrategyTest.java
@@ -0,0 +1,54 @@
+/*
+ * Copyright (c) 2008-2018, Hazelcast, Inc. All Rights Reserved.
+ *
+ * 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.hazelcast.kubernetes;
+
+import com.hazelcast.nio.IOUtil;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.nio.charset.Charset;
+
+public class HazelcastKubernetesDiscoveryStrategyTest {
+
+    @Test
+    public void testReadFileContents()
+            throws IOException {
+        String expectedContents = "Hello, world!\nThis is a test with Unicode ✓.";
+        String testFile = createTestFile(expectedContents);
+        String actualContents = HazelcastKubernetesDiscoveryStrategy.readFileContents(testFile);
+        Assert.assertEquals(expectedContents, actualContents);
+    }
+
+    private static String createTestFile(String expectedContents)
+            throws IOException {
+        File temp = File.createTempFile("test", ".tmp");
+        temp.deleteOnExit();
+        BufferedWriter bufferedWriter = null;
+        try {
+            bufferedWriter = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(temp), Charset.forName("UTF-8")));
+            bufferedWriter.write(expectedContents);
+        } finally {
+            IOUtil.closeResource(bufferedWriter);
+        }
+        return temp.getAbsolutePath();
+    }
+}
diff --git a/src/test/java/com/hazelcast/kubernetes/KubernetesApiEndpointResolverTest.java b/src/test/java/com/hazelcast/kubernetes/KubernetesApiEndpointResolverTest.java
index 6581cd2..26402ca 100644
--- a/src/test/java/com/hazelcast/kubernetes/KubernetesApiEndpointResolverTest.java
+++ b/src/test/java/com/hazelcast/kubernetes/KubernetesApiEndpointResolverTest.java
@@ -20,9 +20,7 @@ import com.hazelcast.kubernetes.KubernetesClient.Endpoints;
 import com.hazelcast.kubernetes.KubernetesClient.EntrypointAddress;
 import com.hazelcast.logging.ILogger;
 import com.hazelcast.logging.NoLogFactory;
-import com.hazelcast.nio.IOUtil;
 import com.hazelcast.spi.discovery.DiscoveryNode;
-import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -31,12 +29,6 @@ import org.powermock.api.mockito.PowerMockito;
 import org.powermock.core.classloader.annotations.PrepareForTest;
 import org.powermock.modules.junit4.PowerMockRunner;
 
-import java.io.BufferedWriter;
-import java.io.File;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.OutputStreamWriter;
-import java.nio.charset.Charset;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -54,8 +46,6 @@ public class KubernetesApiEndpointResolverTest {
     private static final String SERVICE_LABEL_VALUE = "serviceLabelValue";
     private static final Boolean RESOLVE_NOT_READY_ADDRESSES = true;
     private static final String NAMESPACE = "theNamespace";
-    private static final String KUBERNETES_MASTER_URL = "http://bla";
-    private static final String API_TOKEN = "token";
 
     @Mock
     private RetryKubernetesClient client;
@@ -74,7 +64,7 @@ public class KubernetesApiEndpointResolverTest {
         given(client.endpoints(NAMESPACE)).willReturn(endpoints);
 
         KubernetesApiEndpointResolver sut = new KubernetesApiEndpointResolver(LOGGER, null, 0, null, null, NAMESPACE, null,
-                KUBERNETES_MASTER_URL, API_TOKEN);
+                client);
 
         // when
         List<DiscoveryNode> nodes = sut.resolve();
@@ -99,8 +89,7 @@ public class KubernetesApiEndpointResolverTest {
         given(client.endpointsByName(NAMESPACE, SERVICE_NAME)).willReturn(endpoints);
 
         KubernetesApiEndpointResolver sut = new KubernetesApiEndpointResolver(LOGGER, SERVICE_NAME, port, null, null, NAMESPACE,
-                null,
-                KUBERNETES_MASTER_URL, API_TOKEN);
+                null, client);
 
         // when
         List<DiscoveryNode> nodes = sut.resolve();
@@ -117,8 +106,7 @@ public class KubernetesApiEndpointResolverTest {
         given(client.endpointsByLabel(NAMESPACE, SERVICE_LABEL, SERVICE_LABEL_VALUE)).willReturn(endpoints);
 
         KubernetesApiEndpointResolver sut = new KubernetesApiEndpointResolver(LOGGER, null, 0, SERVICE_LABEL, SERVICE_LABEL_VALUE,
-                NAMESPACE,
-                null, KUBERNETES_MASTER_URL, API_TOKEN);
+                NAMESPACE, null, client);
 
         // when
         List<DiscoveryNode> nodes = sut.resolve();
@@ -135,7 +123,7 @@ public class KubernetesApiEndpointResolverTest {
         given(client.endpointsByName(NAMESPACE, SERVICE_NAME)).willReturn(endpoints);
 
         KubernetesApiEndpointResolver sut = new KubernetesApiEndpointResolver(LOGGER, SERVICE_NAME, 0, null, null, NAMESPACE,
-                RESOLVE_NOT_READY_ADDRESSES, KUBERNETES_MASTER_URL, API_TOKEN);
+                RESOLVE_NOT_READY_ADDRESSES, client);
 
         // when
         List<DiscoveryNode> nodes = sut.resolve();
@@ -152,7 +140,7 @@ public class KubernetesApiEndpointResolverTest {
 
         KubernetesApiEndpointResolver sut = new KubernetesApiEndpointResolver(LOGGER, SERVICE_NAME, 0, null, null, NAMESPACE,
                 null,
-                KUBERNETES_MASTER_URL, API_TOKEN);
+                client);
 
         // when
         List<DiscoveryNode> nodes = sut.resolve();
@@ -161,29 +149,6 @@ public class KubernetesApiEndpointResolverTest {
         assertEquals(0, nodes.size());
     }
 
-    @Test
-    public void testReadFileContents()
-            throws IOException {
-        String expectedContents = "Hello, world!\nThis is a test with Unicode ✓.";
-        String testFile = createTestFile(expectedContents);
-        String actualContents = KubernetesApiEndpointResolver.readFileContents(testFile);
-        Assert.assertEquals(expectedContents, actualContents);
-    }
-
-    private static String createTestFile(String expectedContents)
-            throws IOException {
-        File temp = File.createTempFile("test", ".tmp");
-        temp.deleteOnExit();
-        BufferedWriter bufferedWriter = null;
-        try {
-            bufferedWriter = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(temp), Charset.forName("UTF-8")));
-            bufferedWriter.write(expectedContents);
-        } finally {
-            IOUtil.closeResource(bufferedWriter);
-        }
-        return temp.getAbsolutePath();
-    }
-
     private static Endpoints createEndpoints(int customPort) {
         return new Endpoints(asList(createEntrypointAddress(customPort)), Collections.<EntrypointAddress>emptyList());
     }
diff --git a/src/test/java/com/hazelcast/kubernetes/RetryKubernetesClientTest.java b/src/test/java/com/hazelcast/kubernetes/RetryKubernetesClientTest.java
index 2584aee..ee24451 100644
--- a/src/test/java/com/hazelcast/kubernetes/RetryKubernetesClientTest.java
+++ b/src/test/java/com/hazelcast/kubernetes/RetryKubernetesClientTest.java
@@ -36,6 +36,8 @@ public class RetryKubernetesClientTest {
     private static final String SERVICE_LABEL = "someServiceLabel";
     private static final String SERVICE_LABEL_VALUE = "someServiceLabelValue";
     private static final String SERVICE_NAME = "someServiceName";
+    private static final String POD_NAME = "hazelcast-0";
+    private static final String ZONE = "us-central1-a";
 
     private static final Endpoints ENDPOINTS = new Endpoints(null, null);
 
@@ -72,7 +74,7 @@ public class RetryKubernetesClientTest {
                                               .willThrow(KubernetesClientException.class).willReturn(ENDPOINTS);
 
         // when
-        Endpoints result = client.endpoints(NAMESPACE);
+        client.endpoints(NAMESPACE);
 
         // then
         // throws exception
@@ -99,7 +101,7 @@ public class RetryKubernetesClientTest {
                 .willThrow(KubernetesClientException.class).willReturn(ENDPOINTS);
 
         // when
-        Endpoints result = client.endpointsByLabel(NAMESPACE, SERVICE_LABEL, SERVICE_LABEL_VALUE);
+        client.endpointsByLabel(NAMESPACE, SERVICE_LABEL, SERVICE_LABEL_VALUE);
 
         // then
         // throws exception
@@ -126,7 +128,32 @@ public class RetryKubernetesClientTest {
                 .willThrow(KubernetesClientException.class).willReturn(ENDPOINTS);
 
         // when
-        Endpoints result = client.endpointsByName(NAMESPACE, SERVICE_NAME);
+        client.endpointsByName(NAMESPACE, SERVICE_NAME);
+
+        // then
+        // throws exception
+    }
+
+    @Test
+    public void zone() {
+        // given
+        given(mockClient.zone(NAMESPACE, POD_NAME)).willThrow(KubernetesClientException.class).willReturn(ZONE);
+
+        // when
+        String result = client.zone(NAMESPACE, POD_NAME);
+
+        // then
+        assertEquals(ZONE, result);
+    }
+
+    @Test(expected = KubernetesClientException.class)
+    public void zoneRetriesExceeded() {
+        // given
+        given(mockClient.zone(NAMESPACE, POD_NAME)).willThrow(KubernetesClientException.class)
+                                                   .willThrow(KubernetesClientException.class).willReturn(ZONE);
+
+        // when
+        client.zone(NAMESPACE, POD_NAME);
 
         // then
         // throws exception
diff --git a/src/test/java/com/hazelcast/kubernetes/RetryUtilsTest.java b/src/test/java/com/hazelcast/kubernetes/RetryUtilsTest.java
index 4a17895..e2ff628 100644
--- a/src/test/java/com/hazelcast/kubernetes/RetryUtilsTest.java
+++ b/src/test/java/com/hazelcast/kubernetes/RetryUtilsTest.java
@@ -19,10 +19,12 @@ package com.hazelcast.kubernetes;
 import com.hazelcast.core.HazelcastException;
 import org.junit.Test;
 
+import java.util.Collections;
 import java.util.concurrent.Callable;
 
 import static com.hazelcast.kubernetes.RetryUtils.BACKOFF_MULTIPLIER;
 import static com.hazelcast.kubernetes.RetryUtils.INITIAL_BACKOFF_MS;
+import static java.util.Arrays.asList;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.BDDMockito.given;
@@ -33,6 +35,7 @@ import static org.mockito.Mockito.verify;
 public class RetryUtilsTest {
     private static final Integer RETRIES = 1;
     private static final String RESULT = "result string";
+    private static final String NON_RETRYABLE_KEYWORD = "\"reason\":\"Forbidden\"";
 
     private Callable<String> callable = mock(Callable.class);
 
@@ -43,7 +46,7 @@ public class RetryUtilsTest {
         given(callable.call()).willReturn(RESULT);
 
         // when
-        String result = RetryUtils.retry(callable, RETRIES);
+        String result = RetryUtils.retry(callable, RETRIES, Collections.<String>emptyList());
 
         // then
         assertEquals(RESULT, result);
@@ -57,7 +60,7 @@ public class RetryUtilsTest {
         given(callable.call()).willThrow(new RuntimeException()).willReturn(RESULT);
 
         // when
-        String result = RetryUtils.retry(callable, RETRIES);
+        String result = RetryUtils.retry(callable, RETRIES, Collections.<String>emptyList());
 
         // then
         assertEquals(RESULT, result);
@@ -71,7 +74,7 @@ public class RetryUtilsTest {
         given(callable.call()).willThrow(new RuntimeException()).willThrow(new RuntimeException()).willReturn(RESULT);
 
         // when
-        RetryUtils.retry(callable, RETRIES);
+        RetryUtils.retry(callable, RETRIES, Collections.<String>emptyList());
 
         // then
         // throws exception
@@ -84,7 +87,7 @@ public class RetryUtilsTest {
         given(callable.call()).willThrow(new Exception()).willThrow(new Exception()).willReturn(RESULT);
 
         // when
-        RetryUtils.retry(callable, RETRIES);
+        RetryUtils.retry(callable, RETRIES, Collections.<String>emptyList());
 
         // then
         // throws exception
@@ -99,11 +102,43 @@ public class RetryUtilsTest {
 
         // when
         long startTimeMs = System.currentTimeMillis();
-        RetryUtils.retry(callable, 5);
+        RetryUtils.retry(callable, 5, Collections.<String>emptyList());
         long endTimeMs = System.currentTimeMillis();
 
         // then
         assertTrue(twoBackoffIntervalsMs < (endTimeMs - startTimeMs));
     }
 
+    @Test(expected = NonRetryableException.class)
+    public void retryNonRetryableKeyword()
+            throws Exception {
+        // given
+        given(callable.call()).willThrow(new NonRetryableException()).willReturn(RESULT);
+
+        // when
+        RetryUtils.retry(callable, RETRIES, asList(NON_RETRYABLE_KEYWORD));
+
+        // then
+        // throws exception
+    }
+
+    @Test(expected = RuntimeException.class)
+    public void retryNonRetryableKeywordOnCause()
+            throws Exception {
+        // given
+        given(callable.call()).willThrow(new RuntimeException(new NonRetryableException())).willReturn(RESULT);
+
+        // when
+        RetryUtils.retry(callable, RETRIES, asList(NON_RETRYABLE_KEYWORD));
+
+        // then
+        // throws exception
+    }
+
+    private static class NonRetryableException
+            extends RuntimeException {
+        private NonRetryableException() {
+            super(String.format("Message: %s", NON_RETRYABLE_KEYWORD));
+        }
+    }
 }
\ No newline at end of file