<commit_before>package de.fernunihagen.dna.scalephant.distribution.zookeeper;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooKeeper.States;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import de.fernunihagen.dna.scalephant.Const;
import de.fernunihagen.dna.scalephant.ScalephantService;
import de.fernunihagen.dna.scalephant.distribution.DistributionGroupName;
import de.fernunihagen.dna.scalephant.distribution.DistributionRegion;
import de.fernunihagen.dna.scalephant.distribution.DistributionRegionFactory;
import de.fernunihagen.dna.scalephant.distribution.membership.DistributedInstance;
import de.fernunihagen.dna.scalephant.distribution.membership.DistributedInstanceManager;
import de.fernunihagen.dna.scalephant.distribution.membership.event.DistributedInstanceState;

public class ZookeeperClient implements ScalephantService, Watcher {
	
	/**
	 * The list of the zookeeper hosts
	 */
	protected final Collection<String> zookeeperHosts;
	
	/**
	 * The name of the scalephant cluster
	 */
	protected final String clustername;
	
	/**
	 * The zookeeper client instance
	 */
	protected ZooKeeper zookeeper;
	
	/**
	 * The name of the instance
	 */
	protected DistributedInstance instancename = null;
	
	/**
	 * Is the membership observer active?
	 */
	protected volatile boolean membershipObserver = false;
	
	/**
	 * The timeout for the zookeeper session
	 */
	protected final static int DEFAULT_TIMEOUT = 3000;
	

	
	public class NodeNames {
		/**
		 * The prefix for nodes in sequential queues
		 */
		public final static String SEQUENCE_QUEUE_PREFIX = "id-";
		
		/**
		 * Name of the left tree node
		 */
		public final static String NAME_LEFT = "left";
		
		/**
		 * Name of the right tree node
		 */
		public final static String NAME_RIGHT = "right";
		
		/**
		 * Name of the split node
		 */
		public final static String NAME_SPLIT = "split";
		
		/**
		 * Name of the name prefix node
		 */
		public final static String NAME_NAMEPREFIX = "nameprefix";
		
		/**
		 * Name of the name prefix queue
		 */
		public static final String NAME_PREFIXQUEUE = "nameprefixqueue";
	
		/**
		 * Name of the replication node
		 */
		public final static String NAME_REPLICATION = "replication";
		
		/**
		 * Name of the systems node
		 */
		public final static String NAME_SYSTEMS = "systems";
		
		/**
		 * Name of the version node
		 */
		public final static String NAME_VERSION = "version";
		
		/**
		 * Name of the state node
		 */
		public final static String NAME_STATE = "state";
	}
	
	/**
	 * The logger
	 */
	private final static Logger logger = LoggerFactory.getLogger(ZookeeperClient.class);

	public ZookeeperClient(final Collection<String> zookeeperHosts, final String clustername) {
		super();
		this.zookeeperHosts = zookeeperHosts;
		this.clustername = clustername;
		
		// Set the zookeeper instance for self updating data structures
		DistributionRegionFactory.setZookeeperClient(this);
	}

	/**
	 * Connect to zookeeper
	 */
	@Override
	public void init() {
		if(zookeeperHosts == null || zookeeperHosts.isEmpty()) {
			logger.warn("No Zookeeper hosts are defined, not connecting to zookeeper");
			return;
		}
		
		try {
			
			final CountDownLatch connectLatch = new CountDownLatch(1);
			
			zookeeper = new ZooKeeper(generateConnectString(), DEFAULT_TIMEOUT, new Watcher() {
				@Override
				public void process(final WatchedEvent event) {
					if(event.getState() == Watcher.Event.KeeperState.SyncConnected) {
						connectLatch.countDown();
					}
				}
			});
			
			connectLatch.await();
			createDirectoryStructureIfNeeded();
			
			registerInstanceIfNameWasSet();
		} catch (Exception e) {
			logger.warn("Got exception while connecting to zookeeper", e);
		}
	}

	/**
	 * Disconnect from zookeeper
	 */
	@Override
	public void shutdown() {
		if(zookeeper != null) {
			try {
				logger.info("Disconnecting from zookeeper");
				zookeeper.close();
			} catch (InterruptedException e) {
				logger.warn("Got exception while closing zookeeper connection", e);
			}
			zookeeper = null;
		}
	}
	
	/**
	 * Start the membership observer
	 */
	public void startMembershipObserver() {
		membershipObserver = true;
		readMembershipAndRegisterWatch();
	}
	
	public void stopMembershipObserver() {
		membershipObserver = false;
	}
	
	/**
	 * Register a watch on membership changes. A watch is a one-time operation, the watch
	 * is reregistered on each method call.
	 */
	protected void readMembershipAndRegisterWatch() {
		
		if(! membershipObserver) {
			logger.info("Ignore membership event, because observer is not active");
			return;
		}
		
		try {
			final String instancesVersionPath = getInstancesVersionPath(clustername);
			final List<String> instances = zookeeper.getChildren(instancesVersionPath, this);
			final DistributedInstanceManager distributedInstanceManager = DistributedInstanceManager.getInstance();
			
			final Set<DistributedInstance> instanceSet = new HashSet<DistributedInstance>();
			for(final String member : instances) {<commit_msg>Prevent NPE on start membership observer<commit_after>package de.fernunihagen.dna.scalephant.distribution.zookeeper;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooKeeper.States;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import de.fernunihagen.dna.scalephant.Const;
import de.fernunihagen.dna.scalephant.ScalephantService;
import de.fernunihagen.dna.scalephant.distribution.DistributionGroupName;
import de.fernunihagen.dna.scalephant.distribution.DistributionRegion;
import de.fernunihagen.dna.scalephant.distribution.DistributionRegionFactory;
import de.fernunihagen.dna.scalephant.distribution.membership.DistributedInstance;
import de.fernunihagen.dna.scalephant.distribution.membership.DistributedInstanceManager;
import de.fernunihagen.dna.scalephant.distribution.membership.event.DistributedInstanceState;

public class ZookeeperClient implements ScalephantService, Watcher {
	
	/**
	 * The list of the zookeeper hosts
	 */
	protected final Collection<String> zookeeperHosts;
	
	/**
	 * The name of the scalephant cluster
	 */
	protected final String clustername;
	
	/**
	 * The zookeeper client instance
	 */
	protected ZooKeeper zookeeper;
	
	/**
	 * The name of the instance
	 */
	protected DistributedInstance instancename = null;
	
	/**
	 * Is the membership observer active?
	 */
	protected volatile boolean membershipObserver = false;
	
	/**
	 * The timeout for the zookeeper session
	 */
	protected final static int DEFAULT_TIMEOUT = 3000;
	

	
	public class NodeNames {
		/**
		 * The prefix for nodes in sequential queues
		 */
		public final static String SEQUENCE_QUEUE_PREFIX = "id-";
		
		/**
		 * Name of the left tree node
		 */
		public final static String NAME_LEFT = "left";
		
		/**
		 * Name of the right tree node
		 */
		public final static String NAME_RIGHT = "right";
		
		/**
		 * Name of the split node
		 */
		public final static String NAME_SPLIT = "split";
		
		/**
		 * Name of the name prefix node
		 */
		public final static String NAME_NAMEPREFIX = "nameprefix";
		
		/**
		 * Name of the name prefix queue
		 */
		public static final String NAME_PREFIXQUEUE = "nameprefixqueue";
	
		/**
		 * Name of the replication node
		 */
		public final static String NAME_REPLICATION = "replication";
		
		/**
		 * Name of the systems node
		 */
		public final static String NAME_SYSTEMS = "systems";
		
		/**
		 * Name of the version node
		 */
		public final static String NAME_VERSION = "version";
		
		/**
		 * Name of the state node
		 */
		public final static String NAME_STATE = "state";
	}
	
	/**
	 * The logger
	 */
	private final static Logger logger = LoggerFactory.getLogger(ZookeeperClient.class);

	public ZookeeperClient(final Collection<String> zookeeperHosts, final String clustername) {
		super();
		this.zookeeperHosts = zookeeperHosts;
		this.clustername = clustername;
		
		// Set the zookeeper instance for self updating data structures
		DistributionRegionFactory.setZookeeperClient(this);
	}

	/**
	 * Connect to zookeeper
	 */
	@Override
	public void init() {
		if(zookeeperHosts == null || zookeeperHosts.isEmpty()) {
			logger.warn("No Zookeeper hosts are defined, not connecting to zookeeper");
			return;
		}
		
		try {
			
			final CountDownLatch connectLatch = new CountDownLatch(1);
			
			zookeeper = new ZooKeeper(generateConnectString(), DEFAULT_TIMEOUT, new Watcher() {
				@Override
				public void process(final WatchedEvent event) {
					if(event.getState() == Watcher.Event.KeeperState.SyncConnected) {
						connectLatch.countDown();
					}
				}
			});
			
			connectLatch.await();
			createDirectoryStructureIfNeeded();
			
			registerInstanceIfNameWasSet();
		} catch (Exception e) {
			logger.warn("Got exception while connecting to zookeeper", e);
		}
	}

	/**
	 * Disconnect from zookeeper
	 */
	@Override
	public void shutdown() {
		if(zookeeper != null) {
			try {
				logger.info("Disconnecting from zookeeper");
				zookeeper.close();
			} catch (InterruptedException e) {
				logger.warn("Got exception while closing zookeeper connection", e);
			}
			zookeeper = null;
		}
	}
	
	/**
	 * Start the membership observer
	 * @return 
	 */
	public boolean startMembershipObserver() {
		
		if(zookeeper == null) {
			logger.error("startMembershipObserver() called before init()");
			return false;
		}
		
		membershipObserver = true;
		readMembershipAndRegisterWatch();
		
		return true;
	}
	
	public void stopMembershipObserver() {
		membershipObserver = false;
	}
	
	/**
	 * Register a watch on membership changes. A watch is a one-time operation, the watch
	 * is reregistered on each method call.
	 */
	protected void readMembershipAndRegisterWatch() {
		
		if(! membershipObserver) {
			logger.info("Ignore membership event, because observer is not active");
			return;
		}
		
		try {
			final String instancesVersionPath = getInstancesVersionPath(clustername);
			final List<String> instances = zookeeper.getChildren(instancesVersionPath, this);
			final DistributedInstanceManager distributedInstanceManager = DistributedInstanceManager.getInstance();
			
			final Set<DistributedInstance> instanceSet = new HashSet<DistributedInstance>();
			for(final String member : instances) {
{'fraction_non_alphanumeric': 0.09490302210193956, 'fraction_numerical': 0.0012629679747406405, 'mean_word_length': 3.896643109540636, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5567634', 'n_tokens_mistral': 3804, 'n_tokens_neox': 3167, 'n_words': 846}