Repository: hazelcast/hazelcast-kubernetes
Pull request #80 (2018-07-13 13:46:44)

Create our own KubernetesClient (instead of using fabric8)
Create our own Hazelcast KubernetesClient to use Kubernetes REST API. The motivation for this work was to get rid of the dependency to `fabric8.io:kubernetes-client`, which is ~11MB. We actually make only one call to Kubernetes API, so there is no good reason to import the whole library.

Tested on: Minikube, Pivotal PKS, Google Cloud Platform, and OpenShift Online.

Benefits from this change:
- Decreases significantly the size of `hazelcast-kubernetes`: 11MB -> 334KB
- Allows to include `hazelcast-kubernetes` in `hazelcast-all`
- Decreases the Docker image size by 9MB
- Simplifies the code of `ServiceEndpointResolver`

There are also some drawbacks of this change (but I believe they are minor comparing to the benefits):
- We need to maintain `KubernetesClient` (minor, since it's just one class)
- We lose some of the configuration options, which fabric8 provides (e.g. turning off CA Cert verification, etc.) - I believe it's not super important since we allow using `hazelcast-kubernetes` only from inside Kubernetes

Related to: #75


Patch:
diff --git a/pom.xml b/pom.xml
index b18b011..4f31074 100644
--- a/pom.xml
+++ b/pom.xml
@@ -31,11 +31,12 @@
 
     <log4j.version>1.2.12</log4j.version>
     <junit.version>4.12</junit.version>
+    <wiremock.version>2.18.0</wiremock.version>
+    <hamcrest.version>1.3</hamcrest.version>
     <mockito.version>1.10.19</mockito.version>
     <powermock.version>1.7.3</powermock.version>
 
     <hazelcast.version>3.10</hazelcast.version>
-    <kubernetes-client.version>3.1.10</kubernetes-client.version>
     <dnsjava.version>2.1.8</dnsjava.version>
 
     <maven.checkstyle.plugin.version>2.15</maven.checkstyle.plugin.version>
@@ -107,11 +108,6 @@
       <artifactId>hazelcast</artifactId>
       <version>${hazelcast.version}</version>
     </dependency>
-    <dependency>
-      <groupId>io.fabric8</groupId>
-      <artifactId>kubernetes-client</artifactId>
-      <version>${kubernetes-client.version}</version>
-    </dependency>
     <dependency>
       <groupId>dnsjava</groupId>
       <artifactId>dnsjava</artifactId>
@@ -137,6 +133,18 @@
       <scope>test</scope>
       <optional>true</optional>
     </dependency>
+    <dependency>
+      <groupId>com.github.tomakehurst</groupId>
+      <artifactId>wiremock</artifactId>
+      <version>${wiremock.version}</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.hamcrest</groupId>
+      <artifactId>hamcrest-library</artifactId>
+      <version>${hamcrest.version}</version>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>log4j</groupId>
       <artifactId>log4j</artifactId>
@@ -180,8 +188,7 @@
               com.hazelcast.config,com.hazelcast.config.properties,
               com.hazelcast.spi.discovery,com.hazelcast.core,
               com.hazelcast.logging,com.hazelcast.nio, com.hazelcast.util,
-              io.fabric8.kubernetes.api.model,io.fabric8.kubernetes.client.*,
-              org.xbill.DNS,io.fabric8.openshift.client
+              org.xbill.DNS,com.hazelcast.com.eclipsesource.json,javax.net.ssl,javax.security.auth.x500
             </Import-Package>
           </instructions>
         </configuration>
diff --git a/src/main/java/com/hazelcast/kubernetes/KubernetesClient.java b/src/main/java/com/hazelcast/kubernetes/KubernetesClient.java
new file mode 100644
index 0000000..ebca11a
--- /dev/null
+++ b/src/main/java/com/hazelcast/kubernetes/KubernetesClient.java
@@ -0,0 +1,290 @@
+/*
+ * Copyright (c) 2008-2017, 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.com.eclipsesource.json.Json;
+import com.hazelcast.com.eclipsesource.json.JsonArray;
+import com.hazelcast.com.eclipsesource.json.JsonObject;
+import com.hazelcast.com.eclipsesource.json.JsonValue;
+import com.hazelcast.nio.IOUtil;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+
+import javax.net.ssl.HttpsURLConnection;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocketFactory;
+import javax.net.ssl.TrustManagerFactory;
+import java.io.BufferedInputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.security.KeyStore;
+import java.security.cert.Certificate;
+import java.security.cert.CertificateException;
+import java.security.cert.CertificateFactory;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Scanner;
+import java.util.Set;
+
+/**
+ * Responsible for connecting to the Kubernetes API.
+ * <p>
+ * Note: This client should always be used from inside Kubernetes since it depends on the CA Cert file, which exists
+ * in the POD filesystem.
+ *
+ * @see <a href="https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.11/">Kubernetes API</a>
+ */
+class KubernetesClient {
+    private static final int HTTP_OK = 200;
+
+    private final String kubernetesMaster;
+    private final String apiToken;
+
+    KubernetesClient(String kubernetesMaster, String apiToken) {
+        this.kubernetesMaster = kubernetesMaster;
+        this.apiToken = apiToken;
+    }
+
+    /**
+     * Retrieves POD addresses for all services in the given {@code namespace}.
+     *
+     * @param namespace namespace name
+     * @return all POD addresses from the given {@code namespace}
+     * @see <a href="https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.11/#list-143">Kubernetes Endpoint API</a>
+     */
+    Endpoints endpoints(String namespace) {
+        String urlString = String.format("%s/api/v1/namespaces/%s/endpoints", kubernetesMaster, namespace);
+        return parseEndpointsList(callGet(urlString));
+
+    }
+
+    /**
+     * Retrieves POD addresses for all services in the given {@code namespace} filtered by {@code serviceLabel}
+     * and {@code serviceLabelValue}.
+     *
+     * @param namespace         namespace name
+     * @param serviceLabel      label used to filter responses
+     * @param serviceLabelValue label value used to filter responses
+     * @return all POD addresses from the given {@code namespace} filtered by the label
+     * @see <a href="https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.11/#list-143">Kubernetes Endpoint API</a>
+     */
+    Endpoints endpointsByLabel(String namespace, String serviceLabel, String serviceLabelValue) {
+        String param = String.format("labelSelector=%s=%s", serviceLabel, serviceLabelValue);
+        String urlString = String.format("%s/api/v1/namespaces/%s/endpoints?%s", kubernetesMaster, namespace, param);
+        return parseEndpointsList(callGet(urlString));
+    }
+
+    /**
+     * Retrieves POD addresses from the given {@code namespace} and the given {@code endpointName}.
+     *
+     * @param namespace    namespace name
+     * @param endpointName endpoint name
+     * @return all POD addresses from the given {@code namespace} and the given {@code endpointName}
+     * @see <a href="https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.11/#list-143">Kubernetes Endpoint API</a>
+     */
+    Endpoints endpointsByName(String namespace, String endpointName) {
+        String urlString = String.format("%s/api/v1/namespaces/%s/endpoints/%s", kubernetesMaster, namespace, endpointName);
+        JsonObject json = callGet(urlString);
+        return parseEndpoint(json);
+    }
+
+    private JsonObject callGet(String urlString) {
+        HttpURLConnection connection = null;
+        try {
+            URL url = new URL(urlString);
+            connection = (HttpURLConnection) url.openConnection();
+            if (connection instanceof HttpsURLConnection) {
+                ((HttpsURLConnection) connection).setSSLSocketFactory(buildSslSocketFactory());
+            }
+            connection.setRequestMethod("GET");
+            connection.setRequestProperty("Authorization", String.format("Bearer %s", apiToken));
+
+            if (connection.getResponseCode() != HTTP_OK) {
+                throw new KubernetesClientException(String.format("Failure executing: GET at: %s. Message: %s,", urlString,
+                        read(connection.getErrorStream())));
+            }
+            return Json.parse(read(connection.getInputStream())).asObject();
+        } catch (Exception e) {
+            throw new KubernetesClientException("Failure in KubernetesClient", e);
+        } finally {
+            if (connection != null) {
+                connection.disconnect();
+            }
+        }
+    }
+
+    private static String read(InputStream stream) {
+        Scanner scanner = new Scanner(stream, "UTF-8");
+        scanner.useDelimiter("\\Z");
+        return scanner.next();
+    }
+
+    private static Endpoints parseEndpointsList(JsonObject json) {
+        List<EntrypointAddress> addresses = new ArrayList<EntrypointAddress>();
+        List<EntrypointAddress> notReadyAddresses = new ArrayList<EntrypointAddress>();
+
+        for (JsonValue object : toJsonArray(json.get("items"))) {
+            Endpoints endpoints = parseEndpoint(object);
+            addresses.addAll(endpoints.getAddresses());
+            notReadyAddresses.addAll(endpoints.getNotReadyAddresses());
+        }
+
+        return new Endpoints(addresses, notReadyAddresses);
+    }
+
+    private static Endpoints parseEndpoint(JsonValue endpointsJson) {
+        List<EntrypointAddress> addresses = new ArrayList<EntrypointAddress>();
+        List<EntrypointAddress> notReadyAddresses = new ArrayList<EntrypointAddress>();
+
+        for (JsonValue subset : toJsonArray(endpointsJson.asObject().get("subsets"))) {
+            for (JsonValue address : toJsonArray(subset.asObject().get("addresses"))) {
+                addresses.add(parseEntrypointAddress(address));
+            }
+            for (JsonValue notReadyAddress : toJsonArray(subset.asObject().get("notReadyAddresses"))) {
+                notReadyAddresses.add(parseEntrypointAddress(notReadyAddress));
+            }
+        }
+        return new Endpoints(addresses, notReadyAddresses);
+    }
+
+    private static EntrypointAddress parseEntrypointAddress(JsonValue endpointAddressJson) {
+        String ip = endpointAddressJson.asObject().get("ip").asString();
+        Map<String, Object> additionalProperties = parseAdditionalProperties(endpointAddressJson);
+        return new EntrypointAddress(ip, additionalProperties);
+    }
+
+    private static Map<String, Object> parseAdditionalProperties(JsonValue endpointAddressJson) {
+        Set<String> knownFieldNames = new HashSet<String>(Arrays.asList("ip", "nodeName", "targetRef", "hostname"));
+
+        Map<String, Object> result = new HashMap<String, Object>();
+        Iterator<JsonObject.Member> iter = endpointAddressJson.asObject().iterator();
+        while (iter.hasNext()) {
+            JsonObject.Member member = iter.next();
+            if (!knownFieldNames.contains(member.getName())) {
+                result.put(member.getName(), toString(member.getValue()));
+            }
+        }
+        return result;
+    }
+
+    private static JsonArray toJsonArray(JsonValue jsonValue) {
+        if (jsonValue == null || jsonValue.isNull()) {
+            return new JsonArray();
+        } else {
+            return jsonValue.asArray();
+        }
+    }
+
+    private static String toString(JsonValue jsonValue) {
+        if (jsonValue.isString()) {
+            return jsonValue.asString();
+        } else {
+            return jsonValue.toString();
+        }
+    }
+
+    /**
+     * Builds SSL Socket Factory with the public CA Certificate from Kubernetes Master.
+     */
+    private static SSLSocketFactory buildSslSocketFactory() {
+        try {
+            KeyStore keyStore = KeyStore.getInstance(KeyStore.getDefaultType());
+            keyStore.load(null, null);
+            keyStore.setCertificateEntry("ca", generateCertificate());
+
+            TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+            tmf.init(keyStore);
+
+            SSLContext context = SSLContext.getInstance("TLS");
+            context.init(null, tmf.getTrustManagers(), null);
+            return context.getSocketFactory();
+
+        } catch (Exception e) {
+            throw new KubernetesClientException("Failure in generating SSLSocketFactory", e);
+        }
+    }
+
+    /**
+     * Generates CA Certificate from the default CA Cert file (which always exists in the Kubernetes POD).
+     */
+    private static Certificate generateCertificate()
+            throws IOException, CertificateException {
+        InputStream caInput = null;
+        try {
+            CertificateFactory cf = CertificateFactory.getInstance("X.509");
+            caInput = new BufferedInputStream(new FileInputStream(new File(caCertPath())));
+            return cf.generateCertificate(caInput);
+        } finally {
+            IOUtil.closeResource(caInput);
+        }
+    }
+
+    @SuppressFBWarnings("DMI_HARDCODED_ABSOLUTE_FILENAME")
+    private static String caCertPath() {
+        return "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt";
+    }
+
+    /**
+     * Result which stores the information about all addresses.
+     */
+    static final class Endpoints {
+        private final List<EntrypointAddress> addresses;
+        private final List<EntrypointAddress> notReadyAddresses;
+
+        Endpoints(List<EntrypointAddress> addresses, List<EntrypointAddress> notReadyAddresses) {
+            this.addresses = addresses;
+            this.notReadyAddresses = notReadyAddresses;
+        }
+
+        List<EntrypointAddress> getAddresses() {
+            return addresses;
+        }
+
+        List<EntrypointAddress> getNotReadyAddresses() {
+            return notReadyAddresses;
+        }
+    }
+
+    /**
+     * Result which stores the information about a single address.
+     */
+    static final class EntrypointAddress {
+        private final String ip;
+        private final Map<String, Object> additionalProperties;
+
+        EntrypointAddress(String ip, Map<String, Object> additionalProperties) {
+            this.ip = ip;
+            this.additionalProperties = additionalProperties;
+        }
+
+        String getIp() {
+            return ip;
+        }
+
+        Map<String, Object> getAdditionalProperties() {
+            return additionalProperties;
+        }
+    }
+}
diff --git a/src/main/java/com/hazelcast/kubernetes/KubernetesClientException.java b/src/main/java/com/hazelcast/kubernetes/KubernetesClientException.java
new file mode 100644
index 0000000..280c443
--- /dev/null
+++ b/src/main/java/com/hazelcast/kubernetes/KubernetesClientException.java
@@ -0,0 +1,33 @@
+/*
+ * Copyright (c) 2008-2017, 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.core.HazelcastException;
+
+/**
+ * Exception to indicate any issues with {@link KubernetesClient}.
+ */
+class KubernetesClientException
+        extends HazelcastException {
+    KubernetesClientException(String message) {
+        super(message);
+    }
+
+    KubernetesClientException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}
diff --git a/src/main/java/com/hazelcast/kubernetes/ServiceEndpointResolver.java b/src/main/java/com/hazelcast/kubernetes/ServiceEndpointResolver.java
index 3ee6a92..e3af384 100644
--- a/src/main/java/com/hazelcast/kubernetes/ServiceEndpointResolver.java
+++ b/src/main/java/com/hazelcast/kubernetes/ServiceEndpointResolver.java
@@ -16,34 +16,27 @@
 
 package com.hazelcast.kubernetes;
 
+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.Collections;
 import java.util.List;
 import java.util.Map;
 
-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 io.fabric8.kubernetes.api.model.EndpointAddress;
-import io.fabric8.kubernetes.api.model.EndpointSubset;
-import io.fabric8.kubernetes.api.model.Endpoints;
-import io.fabric8.kubernetes.api.model.EndpointsList;
-import io.fabric8.kubernetes.client.Config;
-import io.fabric8.kubernetes.client.ConfigBuilder;
-import io.fabric8.kubernetes.client.DefaultKubernetesClient;
-import io.fabric8.kubernetes.client.KubernetesClient;
-
-class ServiceEndpointResolver extends HazelcastKubernetesDiscoveryStrategy.EndpointResolver {
+class ServiceEndpointResolver
+        extends HazelcastKubernetesDiscoveryStrategy.EndpointResolver {
 
     private final String serviceName;
     private final String serviceLabel;
@@ -72,76 +65,41 @@ class ServiceEndpointResolver extends HazelcastKubernetesDiscoveryStrategy.Endpo
             token = getAccountToken();
         }
         logger.info("Kubernetes Discovery: Bearer Token { " + token + " }");
-        Config config = new ConfigBuilder().withOauthToken(token).withMasterUrl(kubernetesMaster).build();
-        return new DefaultKubernetesClient(config);
+        return new KubernetesClient(kubernetesMaster, token);
     }
 
     @Override
     List<DiscoveryNode> resolve() {
         if (serviceName != null && !serviceName.isEmpty()) {
-            return getSimpleDiscoveryNodes(client.endpoints().inNamespace(namespace).withName(serviceName).get());
+            return getSimpleDiscoveryNodes(client.endpointsByName(namespace, serviceName));
         } else if (serviceLabel != null && !serviceLabel.isEmpty()) {
-            return getDiscoveryNodes(client.endpoints().inNamespace(namespace).withLabel(serviceLabel, serviceLabelValue).list());
+            return getSimpleDiscoveryNodes(client.endpointsByLabel(namespace, serviceLabel, serviceLabelValue));
         }
-        return getNodesByNamespace();
-    }
-
-    private List<DiscoveryNode> getNodesByNamespace() {
-        final EndpointsList endpointsInNamespace = client.endpoints().inNamespace(namespace).list();
-        if (endpointsInNamespace == null) {
-            return Collections.emptyList();
-        }
-        return getDiscoveryNodes(endpointsInNamespace);
-    }
-
-    private List<DiscoveryNode> getDiscoveryNodes(EndpointsList endpointsInNamespace) {
-        if (endpointsInNamespace == null) {
-            return Collections.emptyList();
-        }
-        List<DiscoveryNode> discoveredNodes = new ArrayList<DiscoveryNode>();
-        for (Endpoints endpoints : endpointsInNamespace.getItems()) {
-            discoveredNodes.addAll(getSimpleDiscoveryNodes(endpoints));
-        }
-        return discoveredNodes;
+        return getSimpleDiscoveryNodes(client.endpoints(namespace));
     }
 
     private List<DiscoveryNode> getSimpleDiscoveryNodes(Endpoints endpoints) {
-        if (endpoints == null) {
-            return Collections.emptyList();
-        }
-        List<EndpointSubset> endpointsSubsets = endpoints.getSubsets();
-        if (endpointsSubsets == null) {
-            return Collections.emptyList();
-        }
         List<DiscoveryNode> discoveredNodes = new ArrayList<DiscoveryNode>();
-        for (EndpointSubset endpointSubset : endpointsSubsets) {
-            resolveNotReadyAddresses(discoveredNodes, endpointSubset);
-            resolveAddresses(discoveredNodes, endpointSubset);
-        }
+        resolveNotReadyAddresses(discoveredNodes, endpoints.getNotReadyAddresses());
+        resolveAddresses(discoveredNodes, endpoints.getAddresses());
         return discoveredNodes;
     }
 
-    private void resolveNotReadyAddresses(List<DiscoveryNode> discoveredNodes, EndpointSubset endpointSubset) {
+    private void resolveNotReadyAddresses(List<DiscoveryNode> discoveredNodes, List<EntrypointAddress> notReadyAddresses) {
         if (Boolean.TRUE.equals(resolveNotReadyAddresses)) {
-            for (EndpointAddress endpointAddress : endpointSubset.getNotReadyAddresses()) {
-                addAddress(discoveredNodes, endpointAddress);
-            }
+            resolveAddresses(discoveredNodes, notReadyAddresses);
         }
     }
 
-    private void resolveAddresses(List<DiscoveryNode> discoveredNodes, EndpointSubset endpointSubset) {
-        List<EndpointAddress> endpointAddresses = endpointSubset.getAddresses();
-        if (endpointAddresses == null) {
-            return;
-        }
-        for (EndpointAddress endpointAddress : endpointAddresses) {
-            addAddress(discoveredNodes, endpointAddress);
+    private void resolveAddresses(List<DiscoveryNode> discoveredNodes, List<KubernetesClient.EntrypointAddress> addresses) {
+        for (EntrypointAddress address : addresses) {
+            addAddress(discoveredNodes, address);
         }
     }
 
-    private void addAddress(List<DiscoveryNode> discoveredNodes, EndpointAddress endpointAddress) {
-        Map<String, Object> properties = endpointAddress.getAdditionalProperties();
-        String ip = endpointAddress.getIp();
+    private void addAddress(List<DiscoveryNode> discoveredNodes, EntrypointAddress entrypointAddress) {
+        Map<String, Object> properties = entrypointAddress.getAdditionalProperties();
+        String ip = entrypointAddress.getIp();
         InetAddress inetAddress = mapAddress(ip);
         int port = (this.port > 0) ? this.port : getServicePort(properties);
         Address address = new Address(inetAddress, port);
@@ -151,11 +109,6 @@ class ServiceEndpointResolver extends HazelcastKubernetesDiscoveryStrategy.Endpo
         }
     }
 
-    @Override
-    void destroy() {
-        client.close();
-    }
-
     @SuppressFBWarnings("DMI_HARDCODED_ABSOLUTE_FILENAME")
     private String getAccountToken() {
         return readFileContents("/var/run/secrets/kubernetes.io/serviceaccount/token");


Test patch:
diff --git a/src/test/java/com/hazelcast/kubernetes/DnsEndpointResolverTest.java b/src/test/java/com/hazelcast/kubernetes/DnsEndpointResolverTest.java
index 1534c31..ae3ebff 100644
--- a/src/test/java/com/hazelcast/kubernetes/DnsEndpointResolverTest.java
+++ b/src/test/java/com/hazelcast/kubernetes/DnsEndpointResolverTest.java
@@ -3,13 +3,7 @@ package com.hazelcast.kubernetes;
 import com.hazelcast.logging.ILogger;
 import com.hazelcast.logging.NoLogFactory;
 import com.hazelcast.spi.discovery.DiscoveryNode;
-import io.fabric8.kubernetes.api.model.EndpointAddress;
-import io.fabric8.kubernetes.api.model.EndpointSubset;
-import io.fabric8.kubernetes.api.model.Endpoints;
-import io.fabric8.kubernetes.api.model.EndpointsList;
-import io.fabric8.kubernetes.client.DefaultKubernetesClient;
 import org.junit.Before;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.Mock;
@@ -22,8 +16,6 @@ import org.xbill.DNS.Name;
 import org.xbill.DNS.Record;
 import org.xbill.DNS.SRVRecord;
 
-import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.List;
 
 import static org.junit.Assert.assertEquals;
@@ -39,7 +31,7 @@ public class DnsEndpointResolverTest {
     private static final int SERVICE_DNS_TIMEOUT = 5;
 
     @Mock
-    private DefaultKubernetesClient client;
+    private KubernetesClient client;
 
     @Mock
     private Lookup lookup;
@@ -48,8 +40,9 @@ public class DnsEndpointResolverTest {
     private SRVRecord srvRecord;
 
     @Before
-    public void setup() throws Exception {
-        PowerMockito.whenNew(DefaultKubernetesClient.class).withAnyArguments().thenReturn(client);
+    public void setup()
+            throws Exception {
+        PowerMockito.whenNew(KubernetesClient.class).withAnyArguments().thenReturn(client);
         PowerMockito.whenNew(SRVRecord.class).withAnyArguments().thenReturn(srvRecord);
         when(srvRecord.getTarget()).thenReturn(Name.fromString("127.0.0.1"));
     }
@@ -62,18 +55,23 @@ public class DnsEndpointResolverTest {
     }
 
     @Test
-    public void testValidServiceDns() throws Exception {
+    public void testValidServiceDns()
+            throws Exception {
         testValidServiceDns(0, 5701);
     }
 
     @Test
-    public void testValidServiceDnsWithCustomPort() throws Exception {
+    public void testValidServiceDnsWithCustomPort()
+            throws Exception {
         testValidServiceDns(333, 333);
     }
 
-    private void testValidServiceDns(final int port, final int expectedPort) throws Exception {
-        DnsEndpointResolver endpointResolver = PowerMockito.spy(new DnsEndpointResolver(LOGGER, "hazelcast.com", port, SERVICE_DNS_TIMEOUT));
-        PowerMockito.when(endpointResolver, MemberMatcher.method(DnsEndpointResolver.class, "buildLookup")).withNoArguments().thenReturn(lookup);
+    private void testValidServiceDns(final int port, final int expectedPort)
+            throws Exception {
+        DnsEndpointResolver endpointResolver = PowerMockito
+                .spy(new DnsEndpointResolver(LOGGER, "hazelcast.com", port, SERVICE_DNS_TIMEOUT));
+        PowerMockito.when(endpointResolver, MemberMatcher.method(DnsEndpointResolver.class, "buildLookup")).withNoArguments()
+                    .thenReturn(lookup);
         when(lookup.getResult()).thenReturn(Lookup.SUCCESSFUL);
         when(lookup.run()).thenReturn(getRecords());
         List<DiscoveryNode> nodes = endpointResolver.resolve();
@@ -83,9 +81,12 @@ public class DnsEndpointResolverTest {
     }
 
     @Test
-    public void testDnsFailFlow() throws Exception {
-        DnsEndpointResolver endpointResolver = PowerMockito.spy(new DnsEndpointResolver(LOGGER, "hazelcast.com", 0, SERVICE_DNS_TIMEOUT));
-        PowerMockito.when(endpointResolver, MemberMatcher.method(DnsEndpointResolver.class, "buildLookup")).withNoArguments().thenReturn(lookup);
+    public void testDnsFailFlow()
+            throws Exception {
+        DnsEndpointResolver endpointResolver = PowerMockito
+                .spy(new DnsEndpointResolver(LOGGER, "hazelcast.com", 0, SERVICE_DNS_TIMEOUT));
+        PowerMockito.when(endpointResolver, MemberMatcher.method(DnsEndpointResolver.class, "buildLookup")).withNoArguments()
+                    .thenReturn(lookup);
 
         when(lookup.getResult()).thenReturn(Lookup.HOST_NOT_FOUND);
         when(lookup.run()).thenReturn(getRecords());
diff --git a/src/test/java/com/hazelcast/kubernetes/KubernetesClientTest.java b/src/test/java/com/hazelcast/kubernetes/KubernetesClientTest.java
new file mode 100644
index 0000000..52dc4af
--- /dev/null
+++ b/src/test/java/com/hazelcast/kubernetes/KubernetesClientTest.java
@@ -0,0 +1,320 @@
+/*
+ * Copyright (c) 2008-2017, 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.github.tomakehurst.wiremock.junit.WireMockRule;
+import com.hazelcast.kubernetes.KubernetesClient.Endpoints;
+import org.junit.Rule;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
+import static com.github.tomakehurst.wiremock.client.WireMock.equalTo;
+import static com.github.tomakehurst.wiremock.client.WireMock.get;
+import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlPathMatching;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.junit.Assert.assertTrue;
+
+public class KubernetesClientTest {
+    private static final String KUBERNETES_MASTER_IP = "localhost";
+    private static final int KUBERNETES_MASTER_PORT = 8089;
+    private static final String KUBERNETES_MASTER_URL = String
+            .format("http://%s:%d", KUBERNETES_MASTER_IP, KUBERNETES_MASTER_PORT);
+
+    private static final String TOKEN = "sample-token";
+    private static final String NAMESPACE = "sample-namespace";
+
+    private static final String SAMPLE_ADDRESS_1 = "192.168.0.25";
+    private static final String SAMPLE_ADDRESS_2 = "172.17.0.5";
+    private static final String SAMPLE_NOT_READY_ADDRESS = "172.17.0.6";
+    private static final String SAMPLE_PORT_1 = "5701";
+    private static final String SAMPLE_PORT_2 = "5702";
+    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);
+
+    @Rule
+    public WireMockRule wireMockRule = new WireMockRule(KUBERNETES_MASTER_PORT);
+
+    private final KubernetesClient kubernetesClient = new KubernetesClient(KUBERNETES_MASTER_URL, TOKEN);
+
+    @Test
+    public void endpointsByNamespace() {
+        // given
+        stubFor(get(urlEqualTo(String.format("/api/v1/namespaces/%s/endpoints", NAMESPACE)))
+                .withHeader("Authorization", equalTo(String.format("Bearer %s", TOKEN)))
+                .willReturn(aResponse().withStatus(200).withBody(endpointsListBody())));
+
+        // when
+        Endpoints result = kubernetesClient.endpoints(NAMESPACE);
+
+        // then
+        assertThat(extractIpPort(result.getAddresses()), containsInAnyOrder(SAMPLE_IP_PORT_1, SAMPLE_IP_PORT_2));
+        assertThat(extractIpPort(result.getNotReadyAddresses()), containsInAnyOrder(SAMPLE_NOT_READY_IP));
+    }
+
+    @Test
+    public void endpointsByNamespaceAndLabel() {
+        // given
+        String serviceLabel = "sample-service-label";
+        String serviceLabelValue = "sample-service-label-value";
+        stubFor(get(urlPathMatching(String.format("/api/v1/namespaces/%s/endpoints", NAMESPACE)))
+                .withQueryParam("labelSelector", equalTo(String.format("%s=%s", serviceLabel, serviceLabelValue)))
+                .withHeader("Authorization", equalTo(String.format("Bearer %s", TOKEN)))
+                .willReturn(aResponse().withStatus(200).withBody(endpointsListBody())));
+
+        // when
+        Endpoints result = kubernetesClient.endpointsByLabel(NAMESPACE, serviceLabel, serviceLabelValue);
+
+        // then
+        assertThat(extractIpPort(result.getAddresses()), containsInAnyOrder(SAMPLE_IP_PORT_1, SAMPLE_IP_PORT_2));
+        assertThat(extractIpPort(result.getNotReadyAddresses()), containsInAnyOrder(SAMPLE_NOT_READY_IP));
+
+    }
+
+    @Test
+    public void endpointsByNamespaceAndServiceName() {
+        // given
+        String serviceName = "service-name";
+        stubFor(get(urlPathMatching(String.format("/api/v1/namespaces/%s/endpoints/%s", NAMESPACE, serviceName)))
+                .withHeader("Authorization", equalTo(String.format("Bearer %s", TOKEN)))
+                .willReturn(aResponse().withStatus(200).withBody(endpointsBody())));
+
+        // when
+        Endpoints result = kubernetesClient.endpointsByName(NAMESPACE, serviceName);
+
+        // then
+        assertThat(extractIpPort(result.getAddresses()), containsInAnyOrder(SAMPLE_IP_PORT_1, SAMPLE_IP_PORT_2));
+        assertTrue(result.getNotReadyAddresses().isEmpty());
+    }
+
+    @Test(expected = KubernetesClientException.class)
+    public void forbidden() {
+        // given
+        stubFor(get(urlEqualTo(String.format("/api/v1/namespaces/%s/endpoints", NAMESPACE)))
+                .withHeader("Authorization", equalTo(String.format("Bearer %s", TOKEN)))
+                .willReturn(aResponse().withStatus(501).withBody(forbiddenBody())));
+
+        // when
+        kubernetesClient.endpoints(NAMESPACE);
+    }
+
+    @Test(expected = KubernetesClientException.class)
+    public void malformedResponse() {
+        // given
+        stubFor(get(urlEqualTo(String.format("/api/v1/namespaces/%s/endpoints", NAMESPACE)))
+                .withHeader("Authorization", equalTo(String.format("Bearer %s", TOKEN)))
+                .willReturn(aResponse().withStatus(200).withBody(malformedBody())));
+
+        // when
+        kubernetesClient.endpoints(NAMESPACE);
+    }
+
+    /**
+     * Real response recorded from the Kubernetes API call "/api/v1/namespaces/{namespace}/endpoints".
+     */
+    private static String endpointsListBody() {
+        return String.format(
+                "{" +
+                        "  \"kind\": \"EndpointsList\"," +
+                        "  \"apiVersion\": \"v1\"," +
+                        "  \"metadata\": {" +
+                        "    \"selfLink\": \"/api/v1/namespaces/default/endpoints\"," +
+                        "    \"resourceVersion\": \"8792\"" +
+                        "  }," +
+                        "  \"items\": [" +
+                        "    {" +
+                        "      \"metadata\": {" +
+                        "        \"name\": \"kubernetes\"," +
+                        "        \"namespace\": \"default\"," +
+                        "        \"selfLink\": \"/api/v1/namespaces/default/endpoints/kubernetes\"," +
+                        "        \"uid\": \"01c5aaa4-8411-11e8-abd2-00155d395157\"," +
+                        "        \"resourceVersion\": \"38\"," +
+                        "        \"creationTimestamp\": \"2018-07-10T07:15:21Z\"" +
+                        "      }," +
+                        "      \"subsets\": [" +
+                        "        {" +
+                        "          \"addresses\": [" +
+                        "            {" +
+                        "              \"ip\": \"%s\"," +
+                        "              \"hazelcast-service-port\" :\"%s\"" +
+                        "            }" +
+                        "          ]," +
+                        "          \"ports\": [" +
+                        "            {" +
+                        "              \"name\": \"https\"," +
+                        "              \"port\": 8443," +
+                        "              \"protocol\": \"TCP\"" +
+                        "            }" +
+                        "          ]" +
+                        "        }" +
+                        "      ]" +
+                        "    }," +
+                        "    {" +
+                        "      \"metadata\": {" +
+                        "        \"name\": \"my-hazelcast\"," +
+                        "        \"namespace\": \"default\"," +
+                        "        \"selfLink\": \"/api/v1/namespaces/default/endpoints/my-hazelcast\"," +
+                        "        \"uid\": \"80f7f03d-8425-11e8-abd2-00155d395157\"," +
+                        "        \"resourceVersion\": \"8788\"," +
+                        "        \"creationTimestamp\": \"2018-07-10T09:42:04Z\"," +
+                        "        \"labels\": {" +
+                        "          \"app\": \"hazelcast\"," +
+                        "          \"chart\": \"hazelcast-1.0.0\"," +
+                        "          \"heritage\": \"Tiller\"," +
+                        "          \"release\": \"my-hazelcast\"" +
+                        "        }" +
+                        "      }," +
+                        "      \"subsets\": [" +
+                        "        {" +
+                        "          \"addresses\": [" +
+                        "            {" +
+                        "              \"ip\": \"%s\"," +
+                        "              \"nodeName\": \"minikube\"," +
+                        "              \"hazelcast-service-port\" : %s," +
+                        "              \"targetRef\": {" +
+                        "                \"kind\": \"Pod\"," +
+                        "                \"namespace\": \"default\"," +
+                        "                \"name\": \"my-hazelcast-0\"," +
+                        "                \"uid\": \"80f20bcb-8425-11e8-abd2-00155d395157\"," +
+                        "                \"resourceVersion\": \"8771\"" +
+                        "              }" +
+                        "            }" +
+                        "          ]," +
+                        "          \"notReadyAddresses\": [" +
+                        "            {" +
+                        "              \"ip\": \"%s\"," +
+                        "              \"nodeName\": \"minikube\"," +
+                        "              \"targetRef\": {" +
+                        "                \"kind\": \"Pod\"," +
+                        "                \"namespace\": \"default\"," +
+                        "                \"name\": \"my-hazelcast-1\"," +
+                        "                \"uid\": \"95bd3636-8425-11e8-abd2-00155d395157\"," +
+                        "                \"resourceVersion\": \"8787\"" +
+                        "              }" +
+                        "            }" +
+                        "          ]," +
+                        "          \"ports\": [" +
+                        "            {" +
+                        "              \"name\": \"hzport\"," +
+                        "              \"port\": 5701," +
+                        "              \"protocol\": \"TCP\"" +
+                        "            }" +
+                        "          ]" +
+                        "        }" +
+                        "      ]" +
+                        "    }" +
+                        "  ]" +
+                        "}"
+                , SAMPLE_ADDRESS_1, SAMPLE_PORT_1, SAMPLE_ADDRESS_2, SAMPLE_PORT_2, SAMPLE_NOT_READY_ADDRESS);
+    }
+
+    /**
+     * Real response recorded from the Kubernetes API call "/api/v1/namespaces/{namespace}/endpoints/{endpoint}".
+     */
+    private static String endpointsBody() {
+        return String.format(
+                "{" +
+                        "  \"kind\": \"Endpoints\"," +
+                        "  \"apiVersion\": \"v1\"," +
+                        "  \"metadata\": {" +
+                        "    \"name\": \"my-hazelcast\"," +
+                        "    \"namespace\": \"default\"," +
+                        "    \"selfLink\": \"/api/v1/namespaces/default/endpoints/my-hazelcast\"," +
+                        "    \"uid\": \"deefb354-8443-11e8-abd2-00155d395157\"," +
+                        "    \"resourceVersion\": \"18816\"," +
+                        "    \"creationTimestamp\": \"2018-07-10T13:19:27Z\"," +
+                        "    \"labels\": {" +
+                        "      \"app\": \"hazelcast\"," +
+                        "      \"chart\": \"hazelcast-1.0.0\"," +
+                        "      \"heritage\": \"Tiller\"," +
+                        "      \"release\": \"my-hazelcast\"" +
+                        "    }" +
+                        "  }," +
+                        "  \"subsets\": [" +
+                        "    {" +
+                        "      \"addresses\": [" +
+                        "        {" +
+                        "          \"ip\": \"%s\"," +
+                        "          \"nodeName\": \"minikube\"," +
+                        "          \"hazelcast-service-port\" : %s," +
+                        "          \"targetRef\": {" +
+                        "            \"kind\": \"Pod\"," +
+                        "            \"namespace\": \"default\"," +
+                        "            \"name\": \"my-hazelcast-0\"," +
+                        "            \"uid\": \"def2f426-8443-11e8-abd2-00155d395157\"," +
+                        "            \"resourceVersion\": \"18757\"" +
+                        "          }" +
+                        "        }," +
+                        "        {" +
+                        "          \"ip\": \"%s\"," +
+                        "          \"nodeName\": \"minikube\"," +
+                        "          \"hazelcast-service-port\" : %s," +
+                        "          \"targetRef\": {" +
+                        "            \"kind\": \"Pod\"," +
+                        "            \"namespace\": \"default\"," +
+                        "            \"name\": \"my-hazelcast-1\"," +
+                        "            \"uid\": \"f3b96106-8443-11e8-abd2-00155d395157\"," +
+                        "            \"resourceVersion\": \"18815\"" +
+                        "          }" +
+                        "        }" +
+                        "      ]," +
+                        "      \"ports\": [" +
+                        "        {" +
+                        "          \"name\": \"hzport\"," +
+                        "          \"port\": 5701," +
+                        "          \"protocol\": \"TCP\"" +
+                        "        }" +
+                        "      ]" +
+                        "    }" +
+                        "  ]" +
+                        "}"
+                , SAMPLE_ADDRESS_1, SAMPLE_PORT_1, SAMPLE_ADDRESS_2, SAMPLE_PORT_2);
+    }
+
+    /**
+     * Real response recorded from the Kubernetes API.
+     */
+    private static String forbiddenBody() {
+        return "Forbidden!Configured service account doesn't have access. Service account may have been revoked. "
+                + "endpoints is forbidden: User \"system:serviceaccount:default:default\" cannot list endpoints "
+                + "in the namespace \"default\"";
+    }
+
+    private static String malformedBody() {
+        return "malformed response";
+    }
+
+    private static List<String> extractIpPort(List<KubernetesClient.EntrypointAddress> addresses) {
+        List<String> result = new ArrayList<String>();
+        for (KubernetesClient.EntrypointAddress address : addresses) {
+            String ip = address.getIp();
+            String port = String.valueOf(address.getAdditionalProperties().get("hazelcast-service-port"));
+            result.add(ipPort(ip, port));
+        }
+        return result;
+    }
+
+    private static String ipPort(String ip, String port) {
+        return String.format("%s:%s", ip, port);
+    }
+}
diff --git a/src/test/java/com/hazelcast/kubernetes/KubernetesDiscoveryStrategyFactoryTest.java b/src/test/java/com/hazelcast/kubernetes/KubernetesDiscoveryStrategyFactoryTest.java
index 4b24096..75f7753 100644
--- a/src/test/java/com/hazelcast/kubernetes/KubernetesDiscoveryStrategyFactoryTest.java
+++ b/src/test/java/com/hazelcast/kubernetes/KubernetesDiscoveryStrategyFactoryTest.java
@@ -5,7 +5,6 @@ import com.hazelcast.logging.ILogger;
 import com.hazelcast.logging.NoLogFactory;
 import com.hazelcast.spi.discovery.DiscoveryNode;
 import com.hazelcast.spi.discovery.DiscoveryStrategy;
-import io.fabric8.kubernetes.client.DefaultKubernetesClient;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -31,11 +30,12 @@ public class KubernetesDiscoveryStrategyFactoryTest {
     DiscoveryNode discoveryNode;
 
     @Mock
-    private DefaultKubernetesClient client;
+    private KubernetesClient client;
 
     @Before
-    public void setup() throws Exception {
-        PowerMockito.whenNew(DefaultKubernetesClient.class).withAnyArguments().thenReturn(client);
+    public void setup()
+            throws Exception {
+        PowerMockito.whenNew(KubernetesClient.class).withAnyArguments().thenReturn(client);
     }
 
     @Test
@@ -59,8 +59,8 @@ public class KubernetesDiscoveryStrategyFactoryTest {
         properties.put(KubernetesProperties.KUBERNETES_API_TOKEN.key(), API_TOKEN);
         properties.put(String.valueOf(KubernetesProperties.SERVICE_PORT), 333);
         HazelcastKubernetesDiscoveryStrategyFactory factory = new HazelcastKubernetesDiscoveryStrategyFactory();
-        DiscoveryStrategy strategy   = factory.newDiscoveryStrategy(discoveryNode, LOGGER, properties);
-        assertTrue(strategy instanceof  HazelcastKubernetesDiscoveryStrategy);
+        DiscoveryStrategy strategy = factory.newDiscoveryStrategy(discoveryNode, LOGGER, properties);
+        assertTrue(strategy instanceof HazelcastKubernetesDiscoveryStrategy);
         strategy.start();
         strategy.destroy();
     }
diff --git a/src/test/java/com/hazelcast/kubernetes/ServiceEndpointResolverTest.java b/src/test/java/com/hazelcast/kubernetes/ServiceEndpointResolverTest.java
index 1b7c726..b11e676 100644
--- a/src/test/java/com/hazelcast/kubernetes/ServiceEndpointResolverTest.java
+++ b/src/test/java/com/hazelcast/kubernetes/ServiceEndpointResolverTest.java
@@ -1,15 +1,11 @@
 package com.hazelcast.kubernetes;
 
+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 io.fabric8.kubernetes.api.model.EndpointAddress;
-import io.fabric8.kubernetes.api.model.EndpointSubset;
-import io.fabric8.kubernetes.api.model.Endpoints;
-import io.fabric8.kubernetes.api.model.EndpointsList;
-import io.fabric8.kubernetes.client.DefaultKubernetesClient;
-import io.fabric8.kubernetes.client.dsl.base.BaseOperation;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
@@ -25,19 +21,20 @@ 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;
+import java.util.Map;
 
+import static java.util.Arrays.asList;
 import static org.junit.Assert.assertEquals;
-import static org.mockito.Mockito.when;
+import static org.mockito.BDDMockito.given;
 
 @RunWith(PowerMockRunner.class)
 @PrepareForTest(ServiceEndpointResolver.class)
-/**
- * This test class sent by https://github.com/hazelcast/hazelcast-kubernetes/pull/30
- */
 public class ServiceEndpointResolverTest {
     private static final ILogger LOGGER = new NoLogFactory().getLogger("no");
-    private static final String SERVICE_NAME = "";
+    private static final String SERVICE_NAME = "serviceName";
     private static final String SERVICE_LABEL = "theLabel";
     private static final String SERVICE_LABEL_VALUE = "serviceLabelValue";
     private static final Boolean RESOLVE_NOT_READY_ADDRESSES = true;
@@ -46,127 +43,117 @@ public class ServiceEndpointResolverTest {
     private static final String API_TOKEN = "token";
 
     @Mock
-    private DefaultKubernetesClient client;
-
-    @Mock
-    private BaseOperation endpoints;
-
-    @Mock
-    private BaseOperation inNamespace;
-
-    @Mock
-    private BaseOperation withLabel;
-
-    private EndpointsList nodesInNamespace = new EndpointsList();
-    private EndpointsList nodesWithLabel = new EndpointsList();
+    private KubernetesClient client;
 
     @Before
-    public void setup() throws Exception {
-        PowerMockito.whenNew(DefaultKubernetesClient.class).withAnyArguments().thenReturn(client);
-
-        when(client.endpoints()).thenReturn(endpoints);
-        when(endpoints.inNamespace(NAMESPACE)).thenReturn(inNamespace);
-        when(inNamespace.list()).thenReturn(nodesInNamespace);
-        when(inNamespace.withLabel(SERVICE_LABEL, SERVICE_LABEL_VALUE)).thenReturn(withLabel);
-        when(withLabel.list()).thenReturn(nodesWithLabel);
+    public void setup()
+            throws Exception {
+        PowerMockito.whenNew(KubernetesClient.class).withAnyArguments().thenReturn(client);
     }
 
     @Test
-    public void resolveWithNamespaceAndNoNodeInNamespace() {
-        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, SERVICE_NAME, 0, null, null, NAMESPACE, null, KUBERNETES_MASTER_URL, API_TOKEN);
+    public void resolveWhenNodeInFound() {
+        // given
+        Endpoints endpoints = new Endpoints(Collections.<EntrypointAddress>emptyList(),
+                Collections.<EntrypointAddress>emptyList());
+        given(client.endpoints(NAMESPACE)).willReturn(endpoints);
+
+        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, null, 0, null, null, NAMESPACE, null,
+                KUBERNETES_MASTER_URL, API_TOKEN);
+
+        // when
         List<DiscoveryNode> nodes = sut.resolve();
 
+        // then
         assertEquals(0, nodes.size());
     }
 
     @Test
-    public void resolveWithNamespaceAndNodeInNamespace() {
-        resolveWithNamespaceAndNodeInNamespace(0, 1); // expected port 1 is the kubernetes discovery endpoint port
+    public void resolveWithServiceNameWhenNodeInNamespace() {
+        resolveWithServiceNameWhenNodeInNamespace(0, 1); // expected port 1 is the kubernetes discovery endpoint port
     }
 
     @Test
-    public void resolveWithNamespaceAndNodeInNamespaceAndCustomPort() {
-        resolveWithNamespaceAndNodeInNamespace(333, 333);
+    public void resolveWithServiceNameWhenNodeInNamespaceAndCustomPort() {
+        resolveWithServiceNameWhenNodeInNamespace(333, 333);
     }
 
-    private void resolveWithNamespaceAndNodeInNamespace(final int port, final int expectedPort) {
-        Endpoints discoveryNode = createEndpoints(1);
-        nodesInNamespace.getItems().add(discoveryNode);
+    private void resolveWithServiceNameWhenNodeInNamespace(final int port, final int expectedPort) {
+        // given
+        Endpoints endpoints = createEndpoints(1);
+        given(client.endpointsByName(NAMESPACE, SERVICE_NAME)).willReturn(endpoints);
+
+        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, SERVICE_NAME, port, null, null, NAMESPACE, null,
+                KUBERNETES_MASTER_URL, API_TOKEN);
 
-        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, SERVICE_NAME, port, null, null, NAMESPACE, null, KUBERNETES_MASTER_URL, API_TOKEN);
+        // when
         List<DiscoveryNode> nodes = sut.resolve();
 
+        // then
         assertEquals(1, nodes.size());
         assertEquals(expectedPort, nodes.get(0).getPrivateAddress().getPort());
     }
 
     @Test
-    public void resolveWithNamespaceAndNodeInNamespaceWithNoReadyPods() {
-        Endpoints discoveryNode = createEndpoints(1);
-        discoveryNode.getSubsets().get(0).setAddresses(null);
-        nodesInNamespace.getItems().add(discoveryNode);
-
-        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, SERVICE_NAME, 0, null, null, NAMESPACE, null, KUBERNETES_MASTER_URL, API_TOKEN);
-        List<DiscoveryNode> nodes = sut.resolve();
-
-        assertEquals(0, nodes.size());
-    }
+    public void resolveWithServiceLabelWhenNodeWithServiceLabel() {
+        // given
+        Endpoints endpoints = createEndpoints(2);
+        given(client.endpointsByLabel(NAMESPACE, SERVICE_LABEL, SERVICE_LABEL_VALUE)).willReturn(endpoints);
 
-    @Test
-    public void resolveWithNamespaceAndNodeInNamespaceWithNoSubsets() {
-        Endpoints discoveryNode = createEndpoints(1);
-        discoveryNode.setSubsets(null);
-        nodesInNamespace.getItems().add(discoveryNode);
+        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, null, 0, SERVICE_LABEL, SERVICE_LABEL_VALUE, NAMESPACE,
+                null, KUBERNETES_MASTER_URL, API_TOKEN);
 
-        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, SERVICE_NAME, 0, null, null, NAMESPACE, null, KUBERNETES_MASTER_URL, API_TOKEN);
+        // when
         List<DiscoveryNode> nodes = sut.resolve();
 
-        assertEquals(0, nodes.size());
+        // then
+        assertEquals(1, nodes.size());
+        assertEquals(2, nodes.get(0).getPrivateAddress().getPort());
     }
 
     @Test
-    public void resolveWithServiceLabelAndNodeInNamespace() {
-        nodesInNamespace.getItems().add(createEndpoints(1));
+    public void resolveWithServiceNameWhenNotReadyAddressesAndNotReadyEnabled() {
+        // given
+        Endpoints endpoints = createNotReadyEndpoints(2);
+        given(client.endpointsByName(NAMESPACE, SERVICE_NAME)).willReturn(endpoints);
 
-        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, SERVICE_NAME, 0, SERVICE_LABEL, SERVICE_LABEL_VALUE, NAMESPACE, null, KUBERNETES_MASTER_URL, API_TOKEN);
-        List<DiscoveryNode> nodes = sut.resolve();
-
-        assertEquals(0, nodes.size());
-    }
+        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, SERVICE_NAME, 0, null, null, NAMESPACE,
+                RESOLVE_NOT_READY_ADDRESSES, KUBERNETES_MASTER_URL, API_TOKEN);
 
-    @Test
-    public void resolveWithServiceLabelAndNodeWithServiceLabel() {
-        nodesInNamespace.getItems().add(createEndpoints(1));
-        Endpoints discoveryNode = createEndpoints(2);
-        nodesWithLabel.getItems().add(discoveryNode);
-
-        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, SERVICE_NAME, 0, SERVICE_LABEL, SERVICE_LABEL_VALUE, NAMESPACE, null, KUBERNETES_MASTER_URL, API_TOKEN);
+        // when
         List<DiscoveryNode> nodes = sut.resolve();
 
+        // then
         assertEquals(1, nodes.size());
-        assertEquals(2, nodes.get(0).getPrivateAddress().getPort());
     }
 
     @Test
-    public void resolveWithNamespaceAndNotReadyAddresses() {
-        nodesInNamespace.getItems().add(createEndpoints(1));
-        nodesInNamespace.getItems().add(createNotReadyEndpoints(2));
+    public void resolveWithServiceNameWhenNotReadyAddressesAndNotReadyDisabled() {
+        // given
+        Endpoints endpoints = createNotReadyEndpoints(2);
+        given(client.endpointsByName(NAMESPACE, SERVICE_NAME)).willReturn(endpoints);
+
+        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, SERVICE_NAME, 0, null, null, NAMESPACE, null,
+                KUBERNETES_MASTER_URL, API_TOKEN);
 
-        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, SERVICE_NAME, 0, null, null, NAMESPACE, RESOLVE_NOT_READY_ADDRESSES, KUBERNETES_MASTER_URL, API_TOKEN);
+        // when
         List<DiscoveryNode> nodes = sut.resolve();
 
-        assertEquals(2, nodes.size());
+        // then
+        assertEquals(0, nodes.size());
     }
 
     @Test
-    public void testReadFileContents() throws IOException {
+    public void testReadFileContents()
+            throws IOException {
         String expectedContents = "Hello, world!\nThis is a test with Unicode ✓.";
         String testFile = createTestFile(expectedContents);
         String actualContents = ServiceEndpointResolver.readFileContents(testFile);
         Assert.assertEquals(expectedContents, actualContents);
     }
 
-    private String createTestFile(String expectedContents) throws IOException {
+    private static String createTestFile(String expectedContents)
+            throws IOException {
         File temp = File.createTempFile("test", ".tmp");
         temp.deleteOnExit();
         BufferedWriter bufferedWriter = null;
@@ -179,25 +166,18 @@ public class ServiceEndpointResolverTest {
         return temp.getAbsolutePath();
     }
 
-    private Endpoints createEndpoints(int id) {
-        Endpoints endpoints = new Endpoints();
-        EndpointSubset subset = new EndpointSubset();
-        endpoints.getSubsets().add(subset);
-        EndpointAddress address = new EndpointAddress();
-        subset.getAddresses().add(address);
-        address.setIp("1.1.1.1");
-        address.getAdditionalProperties().put("hazelcast-service-port", String.valueOf(id));
-        return endpoints;
+    private static Endpoints createEndpoints(int customPort) {
+        return new Endpoints(asList(createEntrypointAddress(customPort)), Collections.<EntrypointAddress>emptyList());
+    }
+
+    private static Endpoints createNotReadyEndpoints(int customPort) {
+        return new Endpoints(Collections.<EntrypointAddress>emptyList(), asList(createEntrypointAddress(customPort)));
     }
 
-    private Endpoints createNotReadyEndpoints(int id) {
-        Endpoints endpoints = new Endpoints();
-        EndpointSubset subset = new EndpointSubset();
-        endpoints.getSubsets().add(subset);
-        EndpointAddress address = new EndpointAddress();
-        subset.getNotReadyAddresses().add(address);
-        address.setIp("1.1.1.1");
-        address.getAdditionalProperties().put("hazelcast-service-port", String.valueOf(id));
-        return endpoints;
+    private static EntrypointAddress createEntrypointAddress(int customPort) {
+        String ip = "1.1.1.1";
+        Map<String, Object> additionalProperties = new HashMap<String, Object>();
+        additionalProperties.put("hazelcast-service-port", String.valueOf(customPort));
+        return new EntrypointAddress(ip, additionalProperties);
     }
 }
\ No newline at end of file