ZookeeperDiscoveryStrategy.java

/*
 * Copyright 2020 Hazelcast Inc.
 *
 * Licensed under the Hazelcast Community License (the "License"); you may not use
 * this file except in compliance with the License. You may obtain a copy of the
 * License at
 *
 * http://hazelcast.com/hazelcast-community-license
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OF ANY KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations under the License.
 */

package com.hazelcast.zookeeper;

import com.hazelcast.cluster.Address;
import com.hazelcast.logging.ILogger;
import com.hazelcast.spi.discovery.AbstractDiscoveryStrategy;
import com.hazelcast.spi.discovery.DiscoveryNode;
import com.hazelcast.spi.discovery.SimpleDiscoveryNode;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.x.discovery.ServiceDiscovery;
import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
import org.apache.curator.x.discovery.ServiceInstance;
import org.apache.curator.x.discovery.UriSpec;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;

/**
 * Implementation for Zookeeper Discovery Strategy
 */
public class ZookeeperDiscoveryStrategy extends AbstractDiscoveryStrategy {

    private static final String DEFAULT_PATH = "/discovery/hazelcast";
    private static final String DEFAULT_GROUP = "hazelcast";
    private static final Duration CURATOR_BASE_SLEEP_TIME = Duration.ofSeconds(1);

    private final DiscoveryNode thisNode;
    private final ILogger logger;

    private String group;
    private CuratorFramework client;
    private ServiceDiscovery<Void> serviceDiscovery;
    private ServiceInstance<Void> serviceInstance;

    public ZookeeperDiscoveryStrategy(DiscoveryNode discoveryNode, ILogger logger, Map<String, Comparable> properties) {
        super(logger, properties);
        this.thisNode = discoveryNode;
        this.logger = logger;
    }

    private boolean isMember() {
        return thisNode != null;
    }

    @Override
    public void start() {
        startCuratorClient();

        group = getOrDefault(ZookeeperDiscoveryProperties.GROUP, DEFAULT_GROUP);
        try {
            String path = getOrDefault(ZookeeperDiscoveryProperties.ZOOKEEPER_PATH, DEFAULT_PATH);
            ServiceDiscoveryBuilder<Void> discoveryBuilder = ServiceDiscoveryBuilder.builder(Void.class)
              .basePath(path)
              .client(client);

            if (isMember()) {
                //register members only into zookeeper
                //there no need to register clients
                prepareServiceInstance();
                discoveryBuilder.thisInstance(serviceInstance);
            }
            serviceDiscovery = discoveryBuilder.build();
            serviceDiscovery.start();
        } catch (Exception e) {
            throw new IllegalStateException("Error while talking to ZooKeeper. ", e);
        }
    }

    private void prepareServiceInstance() throws Exception {
        Address privateAddress = thisNode.getPrivateAddress();
        serviceInstance = ServiceInstance.<Void>builder()
          .uriSpec(new UriSpec("{scheme}://{address}:{port}"))
          .address(privateAddress.getHost())
          .port(privateAddress.getPort())
          .name(group)
          .build();
    }

    private void startCuratorClient() {
        String zookeeperUrl = getOrNull(ZookeeperDiscoveryProperties.ZOOKEEPER_URL);
        if (zookeeperUrl == null) {
            throw new IllegalStateException("Zookeeper URL cannot be null.");
        }
        if (logger.isFinestEnabled()) {
            logger.finest("Using " + zookeeperUrl + " as Zookeeper URL");
        }
        client = CuratorFrameworkFactory.newClient(
          zookeeperUrl,
          new ExponentialBackoffRetry((int) CURATOR_BASE_SLEEP_TIME.toMillis(), 3));
        client.start();
    }

    public Iterable<DiscoveryNode> discoverNodes() {
        try {
            Collection<ServiceInstance<Void>> members = serviceDiscovery.queryForInstances(group);
            List<DiscoveryNode> nodes = new ArrayList<>(members.size());
            for (ServiceInstance<Void> serviceInstance : members) {
                Address address = new Address(serviceInstance.getAddress(), serviceInstance.getPort());
                nodes.add(new SimpleDiscoveryNode(address));
            }
            return nodes;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException("Error while talking to ZooKeeper", e);
        } catch (Exception e) {
            throw new IllegalStateException("Error while talking to ZooKeeper", e);
        }
    }

    @Override
    public void destroy() {
        try {
            if (isMember() && serviceDiscovery != null) {
                serviceDiscovery.unregisterService(serviceInstance);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException("Error while talking to ZooKeeper", e);
        } catch (Exception e) {
            throw new IllegalStateException("Error while talking to ZooKeeper", e);
        } finally {
            IOUtils.closeSafely(serviceDiscovery);
            IOUtils.closeSafely(client);
        }
    }
}


















Active mutators

CONDITIONALS_BOUNDARY
EMPTY_RETURNS
FALSE_RETURNS
INCREMENTS
INVERT_NEGS
MATH
NEGATE_CONDITIONALS
NULL_RETURNS
PRIMITIVE_RETURNS
TRUE_RETURNS
VOID_METHOD_CALLS

Tests examined

com.hazelcast.zookeeper.HazelcastIntegrationTest.testIntegration_urlNotConfigured(com.hazelcast.zookeeper.HazelcastIntegrationTest) (92 ms)
com.hazelcast.zookeeper.ClientSmokeTest.testClientCanConnectionToCluster(com.hazelcast.zookeeper.ClientSmokeTest) (8333 ms)
com.hazelcast.zookeeper.HazelcastIntegrationTest.testIntegration(com.hazelcast.zookeeper.HazelcastIntegrationTest) (5545 ms)

Report generated by PIT 1.7.3

1		/*
2		* Copyright 2020 Hazelcast Inc.
3		*
4		* Licensed under the Hazelcast Community License (the "License"); you may not use
5		* this file except in compliance with the License. You may obtain a copy of the
6		* License at
7		*
8		* http://hazelcast.com/hazelcast-community-license
9		*
10		* Unless required by applicable law or agreed to in writing, software
11		* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12		* WARRANTIES OF ANY KIND, either express or implied. See the License for the
13		* specific language governing permissions and limitations under the License.
14		*/
15
16		package com.hazelcast.zookeeper;
17
18		import com.hazelcast.cluster.Address;
19		import com.hazelcast.logging.ILogger;
20		import com.hazelcast.spi.discovery.AbstractDiscoveryStrategy;
21		import com.hazelcast.spi.discovery.DiscoveryNode;
22		import com.hazelcast.spi.discovery.SimpleDiscoveryNode;
23		import org.apache.curator.framework.CuratorFramework;
24		import org.apache.curator.framework.CuratorFrameworkFactory;
25		import org.apache.curator.retry.ExponentialBackoffRetry;
26		import org.apache.curator.x.discovery.ServiceDiscovery;
27		import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
28		import org.apache.curator.x.discovery.ServiceInstance;
29		import org.apache.curator.x.discovery.UriSpec;
30
31		import java.time.Duration;
32		import java.util.ArrayList;
33		import java.util.Collection;
34		import java.util.List;
35		import java.util.Map;
36
37		/**
38		* Implementation for Zookeeper Discovery Strategy
39		*/
40		public class ZookeeperDiscoveryStrategy extends AbstractDiscoveryStrategy {
41
42		private static final String DEFAULT_PATH = "/discovery/hazelcast";
43		private static final String DEFAULT_GROUP = "hazelcast";
44		private static final Duration CURATOR_BASE_SLEEP_TIME = Duration.ofSeconds(1);
45
46		private final DiscoveryNode thisNode;
47		private final ILogger logger;
48
49		private String group;
50		private CuratorFramework client;
51		private ServiceDiscovery<Void> serviceDiscovery;
52		private ServiceInstance<Void> serviceInstance;
53
54		public ZookeeperDiscoveryStrategy(DiscoveryNode discoveryNode, ILogger logger, Map<String, Comparable> properties) {
55		super(logger, properties);
56		this.thisNode = discoveryNode;
57		this.logger = logger;
58		}
59
60		private boolean isMember() {
61	2 1. isMember : negated conditional → KILLED 2. isMember : replaced boolean return with true for com/hazelcast/zookeeper/ZookeeperDiscoveryStrategy::isMember → KILLED	return thisNode != null;
62		}
63
64		@Override
65		public void start() {
66	1 1. start : removed call to com/hazelcast/zookeeper/ZookeeperDiscoveryStrategy::startCuratorClient → KILLED	startCuratorClient();
67
68		group = getOrDefault(ZookeeperDiscoveryProperties.GROUP, DEFAULT_GROUP);
69		try {
70		String path = getOrDefault(ZookeeperDiscoveryProperties.ZOOKEEPER_PATH, DEFAULT_PATH);
71		ServiceDiscoveryBuilder<Void> discoveryBuilder = ServiceDiscoveryBuilder.builder(Void.class)
72		.basePath(path)
73		.client(client);
74
75	1 1. start : negated conditional → KILLED	if (isMember()) {
76		//register members only into zookeeper
77		//there no need to register clients
78	1 1. start : removed call to com/hazelcast/zookeeper/ZookeeperDiscoveryStrategy::prepareServiceInstance → KILLED	prepareServiceInstance();
79		discoveryBuilder.thisInstance(serviceInstance);
80		}
81		serviceDiscovery = discoveryBuilder.build();
82	1 1. start : removed call to org/apache/curator/x/discovery/ServiceDiscovery::start → KILLED	serviceDiscovery.start();
83		} catch (Exception e) {
84		throw new IllegalStateException("Error while talking to ZooKeeper. ", e);
85		}
86		}
87
88		private void prepareServiceInstance() throws Exception {
89		Address privateAddress = thisNode.getPrivateAddress();
90		serviceInstance = ServiceInstance.<Void>builder()
91		.uriSpec(new UriSpec("{scheme}://{address}:{port}"))
92		.address(privateAddress.getHost())
93		.port(privateAddress.getPort())
94		.name(group)
95		.build();
96		}
97
98		private void startCuratorClient() {
99		String zookeeperUrl = getOrNull(ZookeeperDiscoveryProperties.ZOOKEEPER_URL);
100	1 1. startCuratorClient : negated conditional → KILLED	if (zookeeperUrl == null) {
101		throw new IllegalStateException("Zookeeper URL cannot be null.");
102		}
103	1 1. startCuratorClient : negated conditional → SURVIVED	if (logger.isFinestEnabled()) {
104	1 1. startCuratorClient : removed call to com/hazelcast/logging/ILogger::finest → SURVIVED	logger.finest("Using " + zookeeperUrl + " as Zookeeper URL");
105		}
106		client = CuratorFrameworkFactory.newClient(
107		zookeeperUrl,
108		new ExponentialBackoffRetry((int) CURATOR_BASE_SLEEP_TIME.toMillis(), 3));
109	1 1. startCuratorClient : removed call to org/apache/curator/framework/CuratorFramework::start → KILLED	client.start();
110		}
111
112		public Iterable<DiscoveryNode> discoverNodes() {
113		try {
114		Collection<ServiceInstance<Void>> members = serviceDiscovery.queryForInstances(group);
115		List<DiscoveryNode> nodes = new ArrayList<>(members.size());
116		for (ServiceInstance<Void> serviceInstance : members) {
117		Address address = new Address(serviceInstance.getAddress(), serviceInstance.getPort());
118		nodes.add(new SimpleDiscoveryNode(address));
119		}
120	1 1. discoverNodes : replaced return value with null for com/hazelcast/zookeeper/ZookeeperDiscoveryStrategy::discoverNodes → KILLED	return nodes;
121		} catch (InterruptedException e) {
122	1 1. discoverNodes : removed call to java/lang/Thread::interrupt → NO_COVERAGE	Thread.currentThread().interrupt();
123		throw new IllegalStateException("Error while talking to ZooKeeper", e);
124		} catch (Exception e) {
125		throw new IllegalStateException("Error while talking to ZooKeeper", e);
126		}
127		}
128
129		@Override
130		public void destroy() {
131		try {
132	2 1. destroy : negated conditional → SURVIVED 2. destroy : negated conditional → SURVIVED	if (isMember() && serviceDiscovery != null) {
133	1 1. destroy : removed call to org/apache/curator/x/discovery/ServiceDiscovery::unregisterService → SURVIVED	serviceDiscovery.unregisterService(serviceInstance);
134		}
135		} catch (InterruptedException e) {
136	1 1. destroy : removed call to java/lang/Thread::interrupt → NO_COVERAGE	Thread.currentThread().interrupt();
137		throw new IllegalStateException("Error while talking to ZooKeeper", e);
138		} catch (Exception e) {
139		throw new IllegalStateException("Error while talking to ZooKeeper", e);
140		} finally {
141	1 1. destroy : removed call to com/hazelcast/zookeeper/IOUtils::closeSafely → SURVIVED	IOUtils.closeSafely(serviceDiscovery);
142	1 1. destroy : removed call to com/hazelcast/zookeeper/IOUtils::closeSafely → SURVIVED	IOUtils.closeSafely(client);
143		}
144		}
145		}
		Mutations
61		1.1 Location : isMember Killed by : com.hazelcast.zookeeper.HazelcastIntegrationTest.testIntegration(com.hazelcast.zookeeper.HazelcastIntegrationTest) negated conditional → KILLED 2.2 Location : isMember Killed by : com.hazelcast.zookeeper.ClientSmokeTest.testClientCanConnectionToCluster(com.hazelcast.zookeeper.ClientSmokeTest) replaced boolean return with true for com/hazelcast/zookeeper/ZookeeperDiscoveryStrategy::isMember → KILLED
66		1.1 Location : start Killed by : com.hazelcast.zookeeper.HazelcastIntegrationTest.testIntegration_urlNotConfigured(com.hazelcast.zookeeper.HazelcastIntegrationTest) removed call to com/hazelcast/zookeeper/ZookeeperDiscoveryStrategy::startCuratorClient → KILLED
75		1.1 Location : start Killed by : com.hazelcast.zookeeper.HazelcastIntegrationTest.testIntegration(com.hazelcast.zookeeper.HazelcastIntegrationTest) negated conditional → KILLED
78		1.1 Location : start Killed by : com.hazelcast.zookeeper.HazelcastIntegrationTest.testIntegration(com.hazelcast.zookeeper.HazelcastIntegrationTest) removed call to com/hazelcast/zookeeper/ZookeeperDiscoveryStrategy::prepareServiceInstance → KILLED
82		1.1 Location : start Killed by : com.hazelcast.zookeeper.HazelcastIntegrationTest.testIntegration(com.hazelcast.zookeeper.HazelcastIntegrationTest) removed call to org/apache/curator/x/discovery/ServiceDiscovery::start → KILLED
100		1.1 Location : startCuratorClient Killed by : com.hazelcast.zookeeper.HazelcastIntegrationTest.testIntegration_urlNotConfigured(com.hazelcast.zookeeper.HazelcastIntegrationTest) negated conditional → KILLED
103		1.1 Location : startCuratorClient Killed by : none negated conditional → SURVIVED
104		1.1 Location : startCuratorClient Killed by : none removed call to com/hazelcast/logging/ILogger::finest → SURVIVED
109		1.1 Location : startCuratorClient Killed by : com.hazelcast.zookeeper.HazelcastIntegrationTest.testIntegration(com.hazelcast.zookeeper.HazelcastIntegrationTest) removed call to org/apache/curator/framework/CuratorFramework::start → KILLED
120		1.1 Location : discoverNodes Killed by : com.hazelcast.zookeeper.HazelcastIntegrationTest.testIntegration(com.hazelcast.zookeeper.HazelcastIntegrationTest) replaced return value with null for com/hazelcast/zookeeper/ZookeeperDiscoveryStrategy::discoverNodes → KILLED
122		1.1 Location : discoverNodes Killed by : none removed call to java/lang/Thread::interrupt → NO_COVERAGE
132		1.1 Location : destroy Killed by : none negated conditional → SURVIVED 2.2 Location : destroy Killed by : none negated conditional → SURVIVED
133		1.1 Location : destroy Killed by : none removed call to org/apache/curator/x/discovery/ServiceDiscovery::unregisterService → SURVIVED
136		1.1 Location : destroy Killed by : none removed call to java/lang/Thread::interrupt → NO_COVERAGE
141		1.1 Location : destroy Killed by : none removed call to com/hazelcast/zookeeper/IOUtils::closeSafely → SURVIVED
142		1.1 Location : destroy Killed by : none removed call to com/hazelcast/zookeeper/IOUtils::closeSafely → SURVIVED

ZookeeperDiscoveryStrategy.java

Mutations

Active mutators

Tests examined