<commit_before>package io.tilt.minka;

import static com.google.common.collect.Sets.newHashSet;
import static io.tilt.minka.TestUtils.assertDistribution;
import static io.tilt.minka.TestUtils.balancers;
import static io.tilt.minka.TestUtils.buildCluster;
import static io.tilt.minka.TestUtils.createServer;
import static io.tilt.minka.TestUtils.duties;
import static io.tilt.minka.TestUtils.pickAServer;
import static io.tilt.minka.TestUtils.prototypeConfig;
import static io.tilt.minka.TestUtils.shutdownServers;
import static java.lang.Thread.sleep;
import static java.util.Arrays.asList;
import static junit.framework.Assert.assertNotNull;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

import java.util.Arrays;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;

import org.junit.Test;

import io.tilt.minka.api.Config;
import io.tilt.minka.api.Duty;
import io.tilt.minka.api.Pallet;
import io.tilt.minka.core.leader.balancer.Balancer.BalancerMetadata;


public class ClusterResilienceTest {

	private final Pallet p = Pallet.builder("p-huc").build();
	private final Set<Pallet> pallets = newHashSet(p);
	private final Set<Duty> duties = duties(p, 50);
	private final Config proto = prototypeConfig();
	private final long wait = proto.beatToMs(10);

	@Test
	/**
	 * proves there's always a leader
	 * proves the leader learns the distribution status from surviving followers
	 * proves followers redirect their beats to the new leader
	 * proves duties are always distributed at anytime, though in different shards 
	 */
	public void test_jumping_leader() throws Exception {
		proto.getBootstrap().setNamespace("test_jumping_leader");
		int size = 4;
		
		final Set<ServerWhitness> cluster = buildCluster(size, proto, pallets, duties);	
		sleep(wait * 4);
		
		ServerWhitness leader = null;
		for (int i = 0; i < size-1; i++) {
			if (leader==null) {
				leader = pickAServer(cluster, true);
			}
			// prove leaders are being elected in order of apparition
			assertEquals("iteration " + i, String.valueOf(i), leader.getServer().getConfig().getBootstrap().getServerTag());
			leader.getServer().shutdown();
			sleep(wait * 5);
			cluster.remove(leader);
			assertDistribution(cluster, duties);
			// prove there is a leader
			assertNotNull("iteration " + i, (leader = pickAServer(cluster, true))!=null);
		}
		leader.getServer().shutdown();
	}
		
	@Test
	public void test_cluster_falling_apart_all_balancers() throws Exception {
		for (BalancerMetadata meta: balancers()) {
			test_cluster_falling_apart(meta);
		}
	}
	
	
	/**
	 * proves a decaying cluster of all its members but the leader
	 * while all duties still kept distributed
	 */
	public void test_cluster_falling_apart(final BalancerMetadata meta) throws Exception {
		
		proto.getBootstrap().setNamespace("test_cluster_falling_apart");
		final Set<ServerWhitness> set = buildCluster(5, proto, pallets, duties);
		sleep(wait * 3);
		
		final ServerWhitness leader = pickAServer(set, true);
		
		final Iterator<ServerWhitness> it = set.iterator();
		while (it.hasNext()) {
			final ServerWhitness sw = it.next();
			if (!sw.getServer().getClient().isCurrentLeader()) {
				sw.getServer().shutdown();
				it.remove();
				sleep(wait * 5);
				// in anytime all duties must be kept distributed 
				assertDistribution(set, duties);
			}
		}
		
		// finally the only standing shard has them all
		assertDistribution(Arrays.asList(leader), duties);
		leader.getServer().shutdown();
	}
	
	@Test
	/**
	 * proves shards may appear and disappear anytime in the same time
	 * while all duties are still kept distributed
	 * proves proctor and distributor phases are never suspended
	 * proves the leader acknowledges every situation
	 */
	public void test_cluster_highly_unstable() throws Exception {
		proto.getBootstrap().setNamespace("test_cluster_highly_unstable");
		// start with 3 (surely 1st will be leader)
		final ServerWhitness a1 = createServer(proto, duties, pallets, "a1");
		final ServerWhitness a2 = createServer(proto, duties, pallets, "a2");
		final ServerWhitness a3 = createServer(proto, duties, pallets, "a3");
		sleep(wait * 3);
		assertDistribution(asList(a1,a2,a3), duties);
	
		a2.getServer().shutdown();
		// immediately launch a new serie of shards
		final ServerWhitness a4 = createServer(proto, duties, pallets, "a4");
		sleep(wait * 5);
		
		assertDistribution(asList(a1,a3,a4), duties);
		
		a3.getServer().shutdown();
		final ServerWhitness a5 = createServer(proto, duties, pallets, "a5");
		final ServerWhitness a6 = createServer(proto, duties, pallets, "a6");
		sleep(wait * 3);
		
		assertDistribution(asList(a1,a4,a5,a6), duties);
		
		a5.getServer().shutdown();
		a6.getServer().shutdown();
		sleep(wait * 3);		
		
		assertDistribution(asList(a1,a4), duties);

		a4.getServer().shutdown();
		
		sleep(wait * 3);
		
		assertTrue(a1.getServer().getClient().isCurrentLeader());
		assertEquals(duties, a1.getCurrent());
		
		assertDistribution(asList(a1), duties);
		
		a1.getServer().shutdown();
	}
	
	@Test
	public void test_cluster_size_increase_decrease_all_balancers() throws Exception {<commit_msg>WIP test<commit_after>package io.tilt.minka;

import static com.google.common.collect.Sets.newHashSet;
import static io.tilt.minka.TestUtils.assertDistribution;
import static io.tilt.minka.TestUtils.balancers;
import static io.tilt.minka.TestUtils.buildCluster;
import static io.tilt.minka.TestUtils.createServer;
import static io.tilt.minka.TestUtils.duties;
import static io.tilt.minka.TestUtils.pickAServer;
import static io.tilt.minka.TestUtils.prototypeConfig;
import static io.tilt.minka.TestUtils.shutdownServers;
import static java.lang.Thread.sleep;
import static java.util.Arrays.asList;
import static java.util.Collections.emptySet;
import static junit.framework.Assert.assertNotNull;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;

import org.junit.Test;

import io.tilt.minka.api.Config;
import io.tilt.minka.api.Duty;
import io.tilt.minka.api.Pallet;
import io.tilt.minka.core.leader.balancer.Balancer.BalancerMetadata;


public class ClusterResilienceTest {

	private final Pallet p = Pallet.builder("p-huc").build();
	private final Set<Pallet> pallets = newHashSet(p);
	private final Set<Duty> duties = duties(p, 50);
	private final Config proto = prototypeConfig();
	private final long wait = proto.beatToMs(10);

	@Test
	/**
	 * proves there's always a leader
	 * proves the leader learns the distribution status from surviving followers
	 * proves followers redirect their beats to the new leader
	 * proves duties are always distributed at anytime, though in different shards 
	 */
	public void test_jumping_leader() throws Exception {
		proto.getBootstrap().setNamespace("test_jumping_leader");
		int size = 4;
		
		final Set<ServerWhitness> cluster = buildCluster(size, proto, pallets, duties);	
		sleep(wait * 4);
		
		ServerWhitness leader = null;
		for (int i = 0; i < size-1; i++) {
			if (leader==null) {
				leader = pickAServer(cluster, true);
			}
			// prove leaders are being elected in order of apparition
			assertEquals("iteration " + i, String.valueOf(i), leader.getServer().getConfig().getBootstrap().getServerTag());
			leader.getServer().shutdown();
			sleep(wait * 5);
			cluster.remove(leader);
			assertDistribution(cluster, duties);
			// prove there is a leader
			assertNotNull("iteration " + i, (leader = pickAServer(cluster, true))!=null);
		}
		leader.getServer().shutdown();
	}
		
	@Test
	public void test_cluster_falling_apart_all_balancers() throws Exception {
		for (BalancerMetadata meta: balancers()) {
			test_cluster_falling_apart(meta);
		}
	}
	
	
	/**
	 * proves a decaying cluster of all its members but the leader
	 * while all duties still kept distributed
	 */
	public void test_cluster_falling_apart(final BalancerMetadata meta) throws Exception {
		
		proto.getBootstrap().setNamespace("test_cluster_falling_apart");
		final Set<ServerWhitness> set = buildCluster(5, proto, pallets, duties);
		sleep(wait * 3);
		
		final ServerWhitness leader = pickAServer(set, true);
		
		final Iterator<ServerWhitness> it = set.iterator();
		while (it.hasNext()) {
			final ServerWhitness sw = it.next();
			if (!sw.getServer().getClient().isCurrentLeader()) {
				sw.getServer().shutdown();
				it.remove();
				sleep(wait * 5);
				// in anytime all duties must be kept distributed 
				assertDistribution(set, duties);
			}
		}
		
		// finally the only standing shard has them all
		assertDistribution(Arrays.asList(leader), duties);
		leader.getServer().shutdown();
	}
	
	@Test
	/**
	 * proves shards may appear and disappear anytime in the same time
	 * while all duties are still kept distributed
	 * proves proctor and distributor phases are never suspended
	 * proves the leader acknowledges every situation
	 */
	public void test_cluster_highly_unstable() throws Exception {
		proto.getBootstrap().setNamespace("test_cluster_highly_unstable");
		// start with 3 (surely 1st will be leader)
		final ServerWhitness a1 = createServer(proto, duties, pallets, "a1");
		final ServerWhitness a2 = createServer(proto, duties, pallets, "a2");
		final ServerWhitness a3 = createServer(proto, duties, pallets, "a3");
		sleep(wait * 3);
		assertDistribution(asList(a1,a2,a3), duties);
	
		a2.getServer().shutdown();
		// immediately launch a new serie of shards
		final ServerWhitness a4 = createServer(proto, duties, pallets, "a4");
		sleep(wait * 5);
		
		assertDistribution(asList(a1,a3,a4), duties);
		
		a3.getServer().shutdown();
		final ServerWhitness a5 = createServer(proto, duties, pallets, "a5");
		final ServerWhitness a6 = createServer(proto, duties, pallets, "a6");
		sleep(wait * 3);
		
		assertDistribution(asList(a1,a4,a5,a6), duties);
		
		a5.getServer().shutdown();
		a6.getServer().shutdown();
		sleep(wait * 3);		
		
		assertDistribution(asList(a1,a4), duties);

		a4.getServer().shutdown();
		
		sleep(wait * 3);
		
		assertTrue(a1.getServer().getClient().isCurrentLeader());
		assertEquals(duties, a1.getCurrent());
		
		assertDistribution(asList(a1), duties);
		
		a1.getServer().shutdown();
	}
	
	

	@Test
	/**
	 * proves shards may appear and disappear anytime in the same time
	 * while all duties are still kept distributed
	 * proves proctor and distributor phases are never suspended
	 * proves the leader acknowledges every situation
	 */
	public void test_cluster_highly_unstable_surviving() throws Exception {
		proto.getBootstrap().setNamespace("test_cluster_highly_unstable");
		// start with 3 (surely 1st will be leader)
		
		final Set<ServerWhitness> set = buildCluster(5, proto, pallets, emptySet());
		final Collection<Duty> duties = duties(p, 50);
		
		set.iterator().next().getServer().getClient().addAll(duties, null);
		
		sleep(wait * 5);
		assertDistribution(set, duties);
	
		set.iterator().next().getServer().shutdown();
		// immediately launch a new serie of shards
		final ServerWhitness a4 = createServer(proto, emptySet(), pallets, "a4");
		set.add(a4);
		sleep(wait * 5);
		
		assertDistribution(set, duties);
		
		set.iterator().next().getServer().shutdown();
		final ServerWhitness a5 = createServer(proto, emptySet(), pallets, "a5");
		final ServerWhitness a6 = createServer(proto, emptySet(), pallets, "a6");
		sleep(wait * 3);
		
		assertDistribution(asList(a1,a4,a5,a6), duties);
		
		a5.getServer().shutdown();
		a6.getServer().shutdown();
		sleep(wait * 3);		
		
		assertDistribution(asList(a1,a4), duties);

		a4.getServer().shutdown();
		
		sleep(wait * 3);
		
		assertTrue(a1.getServer().getClient().isCurrentLeader());
		assertEquals(duties, a1.getCurrent());
		
		assertDistribution(asList(a1), duties);
		
		a1.getServer().shutdown();
	}
	
	@Test
	public void test_cluster_size_increase_decrease_all_balancers() throws Exception {
{'fraction_non_alphanumeric': 0.12032398129592518, 'fraction_numerical': 0.011189044756179024, 'mean_word_length': 4.6309355900329106, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 15, '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': '4738967', 'n_tokens_mistral': 4212, 'n_tokens_neox': 3689, 'n_words': 941}