/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 org.apache.sling.discovery.oak;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.stream.Collectors;

import org.apache.sling.api.resource.LoginException;
import org.apache.sling.api.resource.ModifiableValueMap;
import org.apache.sling.api.resource.Resource;
import org.apache.sling.api.resource.ResourceResolver;
import org.apache.sling.api.resource.ResourceResolverFactory;
import org.apache.sling.discovery.TopologyEvent;
import org.apache.sling.discovery.base.its.setup.OSGiMock;
import org.apache.sling.discovery.base.its.setup.VirtualInstance;
import org.apache.sling.discovery.base.its.setup.mock.DummyResourceResolverFactory;
import org.apache.sling.discovery.base.its.setup.mock.MockFactory;
import org.apache.sling.discovery.base.its.setup.mock.PropertyProviderImpl;
import org.apache.sling.discovery.commons.providers.BaseTopologyView;
import org.apache.sling.discovery.commons.providers.base.DummyListener;
import org.apache.sling.discovery.commons.providers.spi.base.DescriptorHelper;
import org.apache.sling.discovery.commons.providers.spi.base.DiscoveryLiteConfig;
import org.apache.sling.discovery.commons.providers.spi.base.DiscoveryLiteDescriptor;
import org.apache.sling.discovery.commons.providers.spi.base.DiscoveryLiteDescriptorBuilder;
import org.apache.sling.discovery.commons.providers.spi.base.DummySlingSettingsService;
import org.apache.sling.discovery.commons.providers.spi.base.IdMapService;
import org.apache.sling.discovery.oak.cluster.OakClusterViewService;
import org.apache.sling.discovery.oak.its.setup.OakTestConfig;
import org.apache.sling.discovery.oak.its.setup.OakVirtualInstanceBuilder;
import org.apache.sling.discovery.oak.its.setup.SimulatedLease;
import org.apache.sling.discovery.oak.its.setup.SimulatedLeaseCollection;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.osgi.framework.Constants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class OakDiscoveryServiceTest {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());
    private Log4jErrorCatcher catcher;

    public final class SimpleCommonsConfig implements DiscoveryLiteConfig {

        private long bgIntervalMillis;
        private long bgTimeoutMillis;

        SimpleCommonsConfig(long bgIntervalMillis, long bgTimeoutMillis) {
            this.bgIntervalMillis = bgIntervalMillis;
            this.bgTimeoutMillis = bgTimeoutMillis;
        }

        @Override
        public String getSyncTokenPath() {
            return "/var/synctokens";
        }

        @Override
        public String getIdMapPath() {
            return "/var/idmap";
        }

        @Override
        public long getClusterSyncServiceTimeoutMillis() {
            return bgTimeoutMillis;
        }

        @Override
        public long getClusterSyncServiceIntervalMillis() {
            return bgIntervalMillis;
        }

    }

    @Before
    public void setup() {
        catcher = Log4jErrorCatcher.installErrorCatcher(OakClusterViewService.class);
    }

    @After
    public void teardown() {
        if (catcher != null) {
            final String errorMsg = catcher.getLastErrorMsg();
            if (errorMsg != null) {
                fail("Unexpected error msg : " + errorMsg);
            }
            catcher.uninstall();
            catcher = null;
        }
    }

    @Test
    public void testBindBeforeActivate() throws Exception {
        OakVirtualInstanceBuilder builder =
                (OakVirtualInstanceBuilder) new OakVirtualInstanceBuilder()
                .setDebugName("test")
                .newRepository("/foo/bar", true);
        String slingId = UUID.randomUUID().toString();;
        DiscoveryLiteDescriptorBuilder discoBuilder = new DiscoveryLiteDescriptorBuilder();
        discoBuilder.id("id").me(1).activeIds(1);
        // make sure the discovery-lite descriptor is marked as not final
        // such that the view is not already set before we want it to be
        discoBuilder.setFinal(false);
        DescriptorHelper.setDiscoveryLiteDescriptor(builder.getResourceResolverFactory(),
                discoBuilder);
        IdMapService idMapService = IdMapService.testConstructor(new SimpleCommonsConfig(1000, -1), new DummySlingSettingsService(slingId), builder.getResourceResolverFactory());
        assertTrue(idMapService.waitForInit(2000));
        OakDiscoveryService discoveryService = (OakDiscoveryService) builder.getDiscoverService();
        assertNotNull(discoveryService);
        DummyListener listener = new DummyListener();
        for(int i=0; i<100; i++) {
            discoveryService.bindTopologyEventListener(listener);
            discoveryService.unbindTopologyEventListener(listener);
        }
        discoveryService.bindTopologyEventListener(listener);
        assertEquals(0, listener.countEvents());
        discoveryService.activate(null);
        assertEquals(0, listener.countEvents());
        // some more confusion...
        discoveryService.unbindTopologyEventListener(listener);
        discoveryService.bindTopologyEventListener(listener);
        // only set the final flag now - this makes sure that handlePotentialTopologyChange
        // will actually detect a valid new, different view and send out an event -
        // exactly as we want to
        discoBuilder.setFinal(true);
        DescriptorHelper.setDiscoveryLiteDescriptor(builder.getResourceResolverFactory(),
                discoBuilder);
        // SLING-6924 : need to simulate a OakViewChecker.activate to trigger resetLeaderElectionId
        // otherwise no TOPOLOGY_INIT will be generated as without a leaderElectionId we now
        // consider a view as NO_ESTABLISHED_VIEW
        OSGiMock.activate(builder.getViewChecker());
        discoveryService.checkForTopologyChange();
        assertEquals(0, discoveryService.getViewStateManager().waitForAsyncEvents(2000));
        assertEquals(1, listener.countEvents());
        discoveryService.unbindTopologyEventListener(listener);
        assertEquals(1, listener.countEvents());
        discoveryService.bindTopologyEventListener(listener);
        assertEquals(0, discoveryService.getViewStateManager().waitForAsyncEvents(2000));
        assertEquals(2, listener.countEvents()); // should now have gotten an INIT too
    }

    @Test
    public void testDescriptorSeqNumChange() throws Exception {
        logger.info("testDescriptorSeqNumChange: start");
        OakVirtualInstanceBuilder builder1 =
                (OakVirtualInstanceBuilder) new OakVirtualInstanceBuilder()
                .setDebugName("instance1")
                .newRepository("/foo/barry/foo/", true)
                .setConnectorPingInterval(999)
                .setConnectorPingTimeout(999);
        builder1.getConfig().setSuppressPartiallyStartedInstance(true);
;
        VirtualInstance instance1 = builder1.build();
        OakVirtualInstanceBuilder builder2 =
                (OakVirtualInstanceBuilder) new OakVirtualInstanceBuilder()
                .setDebugName("instance2")
                .useRepositoryOf(instance1)
                .setConnectorPingInterval(999)
                .setConnectorPingTimeout(999);
        builder2.getConfig().setSuppressPartiallyStartedInstance(true);
        VirtualInstance instance2 = builder2.build();
        logger.info("testDescriptorSeqNumChange: created both instances, binding listener...");

        DummyListener listener = new DummyListener();
        OakDiscoveryService discoveryService = (OakDiscoveryService) instance1.getDiscoveryService();
        discoveryService.bindTopologyEventListener(listener);

        logger.info("testDescriptorSeqNumChange: waiting 2sec, listener should not get anything yet");
        assertEquals(0, discoveryService.getViewStateManager().waitForAsyncEvents(2000));
        assertEquals(0, listener.countEvents());

        logger.info("testDescriptorSeqNumChange: issuing 2 heartbeats with each instance should let the topology get established");
        instance1.heartbeatsAndCheckView();
        instance2.heartbeatsAndCheckView();
        instance1.heartbeatsAndCheckView();
        instance2.heartbeatsAndCheckView();

        logger.info("testDescriptorSeqNumChange: listener should get an event within 2sec from now at latest");
        assertEquals(0, discoveryService.getViewStateManager().waitForAsyncEvents(2000));
        assertEquals(1, listener.countEvents());

        ResourceResolverFactory factory = instance1.getResourceResolverFactory();
        @SuppressWarnings("unused")
        ResourceResolver resolver = factory.getServiceResourceResolver(null);

        instance1.heartbeatsAndCheckView();
        assertEquals(0, discoveryService.getViewStateManager().waitForAsyncEvents(2000));
        assertEquals(1, listener.countEvents());

        // increment the seqNum by 2 - simulating a coming and going instance
        // while we were sleeping
        SimulatedLeaseCollection c = builder1.getSimulatedLeaseCollection();
        c.incSeqNum(2);
        logger.info("testDescriptorSeqNumChange: incremented seqnum by 2 - issuing another heartbeat should trigger a topology change");
        instance1.heartbeatsAndCheckView();

        // due to the nature of the syncService/minEventDelay we now explicitly first sleep 2sec before waiting for async events for another 2sec
        logger.info("testDescriptorSeqNumChange: sleeping 2sec for topology change to happen");
        Thread.sleep(2000);
        logger.info("testDescriptorSeqNumChange: ensuring no async events are still in the pipe - for another 2sec");
        assertEquals(0, discoveryService.getViewStateManager().waitForAsyncEvents(2000));
        logger.info("testDescriptorSeqNumChange: now listener should have received 3 events, it got: "+listener.countEvents());
        assertEquals(3, listener.countEvents());
    }

    @Test
    public void testNotYetInitializedLeaderElectionid() throws Exception {
        logger.info("testNotYetInitializedLeaderElectionid: start");
        OakVirtualInstanceBuilder builder1 =
                (OakVirtualInstanceBuilder) new OakVirtualInstanceBuilder()
                .setDebugName("instance")
                .newRepository("/foo/barrx/foo/", true)
                .setConnectorPingInterval(999)
                .setConnectorPingTimeout(999);
        builder1.getConfig().setSuppressPartiallyStartedInstance(true);
        VirtualInstance instance1 = builder1.build();
        logger.info("testNotYetInitializedLeaderElectionid: created 1 instance, binding listener...");

        DummyListener listener = new DummyListener();
        OakDiscoveryService discoveryService = (OakDiscoveryService) instance1.getDiscoveryService();
        discoveryService.bindTopologyEventListener(listener);

        logger.info("testNotYetInitializedLeaderElectionid: waiting 2sec, listener should not get anything yet");
        assertEquals(0, discoveryService.getViewStateManager().waitForAsyncEvents(2000));
        assertEquals(0, listener.countEvents());

        logger.info("testNotYetInitializedLeaderElectionid: issuing 2 heartbeats with each instance should let the topology get established");
        instance1.heartbeatsAndCheckView();
        instance1.heartbeatsAndCheckView();

        logger.info("testNotYetInitializedLeaderElectionid: listener should get an event within 2sec from now at latest");
        assertEquals(0, discoveryService.getViewStateManager().waitForAsyncEvents(2000));
        assertEquals(1, listener.countEvents());

        SimulatedLeaseCollection c = builder1.getSimulatedLeaseCollection();
        String secondSlingId = UUID.randomUUID().toString();
        final SimulatedLease newIncomingInstance = new SimulatedLease(instance1.getResourceResolverFactory(), c, secondSlingId);
        c.hooked(newIncomingInstance);
        c.incSeqNum(1);
        newIncomingInstance.updateLeaseAndDescriptor(new OakTestConfig());
        
        logger.info("testNotYetInitializedLeaderElectionid: issuing another 2 heartbeats");
        instance1.heartbeatsAndCheckView();
        instance1.heartbeatsAndCheckView();
        
        // there are different properties that an instance must set in the repository such that it finally becomes visible.
        // these include:
        // 1) idmap : it must map the oak id to sling id
        // 2) node named after its own slingId under /var/discovery/oak/clusterInstances/<slingId>
        // 3) store the leaderElectionId under /var/discovery/oak/clusterInstances/<slingId>
        // in all 3 cases the code must work fine if that node/property doesn't exist
        // and that's exactly what we're testing here.


        // initially not even the idmap is updated, so we're stuck with TOPOLOGY_CHANGING

        // due to the nature of the syncService/minEventDelay we now explicitly first sleep 2sec before waiting for async events for another 2sec
        logger.info("testNotYetInitializedLeaderElectionid: sleeping 2sec for topology change to happen");
        Thread.sleep(2000);
        logger.info("testNotYetInitializedLeaderElectionid: ensuring no async events are still in the pipe - for another 2sec");
        assertEquals(0, discoveryService.getViewStateManager().waitForAsyncEvents(2000));
        logger.info("testNotYetInitializedLeaderElectionid: now listener should have received 2 events, INIT and CHANGING, it got: "+listener.countEvents());
        assertEquals(2, listener.countEvents());
        List<TopologyEvent> events = listener.getEvents();
        assertEquals(TopologyEvent.Type.TOPOLOGY_INIT, events.get(0).getType());
        assertEquals(TopologyEvent.Type.TOPOLOGY_CHANGING, events.get(1).getType());
        
        // let's update the idmap first then
        DummyResourceResolverFactory factory1 = (DummyResourceResolverFactory) instance1.getResourceResolverFactory();
        ResourceResolverFactory factory2 = MockFactory.mockResourceResolverFactory(factory1.getSlingRepository());

        ResourceResolver resourceResolver = getResourceResolver(instance1.getResourceResolverFactory());
        DiscoveryLiteDescriptor descriptor =
                DiscoveryLiteDescriptor.getDescriptorFrom(resourceResolver);
        resourceResolver.close();
        
        DiscoveryLiteDescriptorBuilder dlb = prefill(descriptor);
        dlb.me(2);
        DescriptorHelper.setDiscoveryLiteDescriptor(factory2, dlb);

        @SuppressWarnings("unused")
        IdMapService secondIdMapService = IdMapService.testConstructor((DiscoveryLiteConfig) builder1.getConnectorConfig(), new DummySlingSettingsService(secondSlingId), factory2);
        
        instance1.heartbeatsAndCheckView();
        instance1.heartbeatsAndCheckView();
        Thread.sleep(2000);
        assertEquals(2, listener.countEvents());
        
        
        // now let's add the /var/discovery/oak/clusterInstances/<slingId> node
        resourceResolver = getResourceResolver(factory2);
        Resource clusterInstancesRes = resourceResolver.getResource(builder1.getConnectorConfig().getClusterInstancesPath());
        assertNull(clusterInstancesRes.getChild(secondSlingId));
        resourceResolver.create(clusterInstancesRes, secondSlingId, null);
        resourceResolver.commit();
        assertNotNull(clusterInstancesRes.getChild(secondSlingId));
        resourceResolver.close();

        instance1.heartbeatsAndCheckView();
        instance1.heartbeatsAndCheckView();
        Thread.sleep(2000);
        assertEquals(2, listener.countEvents());

        // now let's add the leaderElectionId
        resourceResolver = getResourceResolver(factory2);
        Resource instanceResource = resourceResolver.getResource(builder1.getConnectorConfig().getClusterInstancesPath() + "/" + secondSlingId);
        assertNotNull(instanceResource);
        instanceResource.adaptTo(ModifiableValueMap.class).put("leaderElectionId", "0");
        resourceResolver.commit();
        resourceResolver.close();
        
        instance1.heartbeatsAndCheckView();
        instance1.heartbeatsAndCheckView();
        Thread.sleep(2000);
        assertEquals(3, listener.countEvents());
        assertEquals(TopologyEvent.Type.TOPOLOGY_CHANGED, events.get(2).getType());
    }

    private DiscoveryLiteDescriptorBuilder prefill(DiscoveryLiteDescriptor d) throws Exception {
        DiscoveryLiteDescriptorBuilder b = new DiscoveryLiteDescriptorBuilder();
        b.setFinal(true);
        long seqnum = d.getSeqNum();
        b.seq((int) seqnum);
        b.activeIds(box(d.getActiveIds()));
        b.deactivatingIds(box(d.getDeactivatingIds()));
        b.me(d.getMyId());
        b.id(d.getViewId());
        return b;
    }

    private Integer[] box(final int[] ids) {
        return Arrays.stream( ids ).boxed().collect( Collectors.toList() ).toArray(new Integer[ids.length]);
    }
    
    private ResourceResolver getResourceResolver(ResourceResolverFactory resourceResolverFactory) throws LoginException {
        return resourceResolverFactory.getServiceResourceResolver(null);
    }

    @Test
    public void testInvertedLeaderElectionOrder() throws Exception {
        logger.info("testInvertedLeaderElectionOrder: start");
        // instance1 should be leader
        BaseTopologyView lastView = doTestInvertedLeaderElectionOrder(20, 10, false);
        assertTrue(lastView.getLocalInstance().isLeader());
        lastView = doTestInvertedLeaderElectionOrder(20, 10, true);
        assertTrue(lastView.getLocalInstance().isLeader());
        // instance2 should be leader
        lastView = doTestInvertedLeaderElectionOrder(10, 20, false);
        assertFalse(lastView.getLocalInstance().isLeader());
        lastView = doTestInvertedLeaderElectionOrder(10, 20, true);
        assertFalse(lastView.getLocalInstance().isLeader());
        // instance1 should be leader
        lastView = doTestInvertedLeaderElectionOrder(10, 10, false);
        assertTrue(lastView.getLocalInstance().isLeader());
        lastView = doTestInvertedLeaderElectionOrder(10, 10, true);
        assertTrue(lastView.getLocalInstance().isLeader());
    }

    private BaseTopologyView doTestInvertedLeaderElectionOrder(final int instance1Prefix, final int instance2Prefix, boolean syncTokenEnabled)
            throws Exception {
        OakVirtualInstanceBuilder builder1 =
                (OakVirtualInstanceBuilder) new OakVirtualInstanceBuilder()
                .setDebugName("instance1")
                .newRepository("/foo/barrio/foo/", true)
                .setConnectorPingInterval(999)
                .setConnectorPingTimeout(999);
        builder1.getConfig().leaderElectionPrefix = instance1Prefix;
        builder1.getConfig().invertLeaderElectionPrefixOrder = true;
        builder1.getConfig().setSyncTokenEnabled(syncTokenEnabled);
        builder1.getConfig().setJoinerDelaySeconds(0);
        builder1.getConfig().setSuppressPartiallyStartedInstance(true);
        VirtualInstance instance1 = builder1.build();
        OakVirtualInstanceBuilder builder2 =
                (OakVirtualInstanceBuilder) new OakVirtualInstanceBuilder()
                .setDebugName("instance2")
                .useRepositoryOf(instance1)
                .setConnectorPingInterval(999)
                .setConnectorPingTimeout(999);
        builder2.getConfig().leaderElectionPrefix = instance2Prefix;
        builder2.getConfig().invertLeaderElectionPrefixOrder = true;
        builder2.getConfig().setSyncTokenEnabled(syncTokenEnabled);
        builder2.getConfig().setJoinerDelaySeconds(0);
        builder2.getConfig().setSuppressPartiallyStartedInstance(true);
        VirtualInstance instance2 = builder2.build();
    
        DummyListener listener = new DummyListener();
        OakDiscoveryService discoveryService = (OakDiscoveryService) instance1.getDiscoveryService();
        discoveryService.bindTopologyEventListener(listener);
    
        instance2.heartbeatsAndCheckView();
        instance1.heartbeatsAndCheckView();
        instance2.heartbeatsAndCheckView();
        instance1.heartbeatsAndCheckView();
    
        assertEquals(0, discoveryService.getViewStateManager().waitForAsyncEvents(2000));
        assertEquals(1, listener.countEvents());
        instance1.stop();
        instance2.stop();
        return listener.getLastView();
    }

    @Test
    public void testPropertyProviderRegistrations() throws Exception{
        OakVirtualInstanceBuilder builder =
                (OakVirtualInstanceBuilder) new OakVirtualInstanceBuilder()
                .setDebugName("instance1")
                .newRepository("/foo/barrio/foo/", true)
                .setConnectorPingInterval(999)
                .setConnectorPingTimeout(999);
        builder.getConfig().setSuppressPartiallyStartedInstance(true);
        VirtualInstance instance = builder.build();
        OakDiscoveryService discoveryService = (OakDiscoveryService) instance.getDiscoveryService();
        discoveryService.bindPropertyProvider(null, null);
        discoveryService.unbindPropertyProvider(null, null);
        discoveryService.updatedPropertyProvider(null, null);

        PropertyProviderImpl p = new PropertyProviderImpl();
        discoveryService.bindPropertyProvider(p, null);
        discoveryService.unbindPropertyProvider(p, null);
        discoveryService.updatedPropertyProvider(p, null);

        Map<String, Object> m = new HashMap<>();
        m.put(Constants.SERVICE_ID, 42L);
        discoveryService.bindPropertyProvider(p, m);
        discoveryService.unbindPropertyProvider(p, m);
        discoveryService.updatedPropertyProvider(p, m);
    }

    @Test
    public void simpleNewJoinerTest() throws Exception {
        doSimpleNewJoinerTest(0);
    }

    @Test
    public void simpleNewJoinerTestWithDelay() throws Exception {
        doSimpleNewJoinerTest(2);
    }

    private boolean waitForEquals(DummyListener listener, int expected, long maxWait) {
        final long timeout = System.currentTimeMillis() + maxWait;
        while(System.currentTimeMillis() < timeout) {
            if (listener.countEvents() == expected) {
                return true;
            }
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                // this empty catch is okay
            }
        }
        final int actual = listener.countEvents();
        if (actual == expected) {
            return true;
        }
        fail("listener did not get expected number of events, expected: " + expected + ", got : " + actual);
        // never reached:
        return false;
    }

    private void doSimpleNewJoinerTest(int joinerDelaySeconds) throws Exception {
        OakVirtualInstanceBuilder builder1 =
                (OakVirtualInstanceBuilder) new OakVirtualInstanceBuilder()
                .setDebugName("instance1")
                .newRepository("/foo/barrio/foo/", true)
                .setConnectorPingInterval(999)
                .setConnectorPingTimeout(999);
        builder1.getConfig().setSyncTokenEnabled(true);
        builder1.getConfig().setJoinerDelaySeconds(joinerDelaySeconds);
        builder1.getConfig().setSuppressPartiallyStartedInstance(true);
        VirtualInstance instance1 = builder1.build();
        DummyListener listener1 = new DummyListener();
        OakDiscoveryService discoveryService1 = (OakDiscoveryService) instance1.getDiscoveryService();
        discoveryService1.bindTopologyEventListener(listener1);

        instance1.heartbeatsAndCheckView();
        instance1.heartbeatsAndCheckView();

        assertEquals(0, discoveryService1.getViewStateManager().waitForAsyncEvents(2000));
        Thread.sleep(1000);
        assertEquals(1, listener1.countEvents());

        OakVirtualInstanceBuilder builder2 =
                (OakVirtualInstanceBuilder) new OakVirtualInstanceBuilder()
                .setDebugName("instance2")
                .useRepositoryOf(instance1)
                .setConnectorPingInterval(999)
                .setConnectorPingTimeout(999);
        builder2.getConfig().setSuppressPartiallyStartedInstance(true);
        builder2.getConfig().setJoinerDelaySeconds(joinerDelaySeconds);

        // while we started instance 2, we don't let it heartbeat yet - just a lease update
        // this simulates a situation where oak is up, but sling-repository not yet
        builder2.updateLease();

        // that way it shouldn't cause TOPOLOGY_CHANGING at instance 1 yet
        instance1.heartbeatsAndCheckView();
        instance1.heartbeatsAndCheckView();

        assertEquals(0, discoveryService1.getViewStateManager().waitForAsyncEvents(2000));
        assertEquals(1, listener1.countEvents());

        instance1.heartbeatsAndCheckView();

        builder2.getConfig().setSyncTokenEnabled(true);
        VirtualInstance instance2 = builder2.build();
        DummyListener listener2 = new DummyListener();
        OakDiscoveryService discoveryService2 = (OakDiscoveryService) instance2.getDiscoveryService();
        discoveryService2.bindTopologyEventListener(listener2);

        instance2.heartbeatsAndCheckView();
        instance1.heartbeatsAndCheckView();
        instance2.heartbeatsAndCheckView();

        Thread.sleep((joinerDelaySeconds + 2) * 1000);

        waitForEquals(listener1, 3, 1000);
        waitForEquals(listener2, 1, 3000); // had a flaky failure so giving it 3 sec

        OakVirtualInstanceBuilder builder3 =
                (OakVirtualInstanceBuilder) new OakVirtualInstanceBuilder()
                .setDebugName("instance3")
                .useRepositoryOf(instance1)
                .setConnectorPingInterval(999)
                .setConnectorPingTimeout(999);
        builder3.getConfig().setJoinerDelaySeconds(joinerDelaySeconds);
        builder3.getConfig().setSuppressPartiallyStartedInstance(true);
        builder3.updateLease();

        instance1.heartbeatsAndCheckView();
        instance2.heartbeatsAndCheckView();
        for(int i=0; i<5; i++) {
            Thread.sleep(300);
            instance1.heartbeatsAndCheckView();
            instance2.heartbeatsAndCheckView();
            assertEquals(3, listener1.countEvents());
            assertEquals(1, listener2.countEvents());
        }

        builder3.getConfig().setSyncTokenEnabled(true);
        VirtualInstance instance3 = builder3.build();
        DummyListener listener3 = new DummyListener();
        OakDiscoveryService discoveryService3 = (OakDiscoveryService) instance3.getDiscoveryService();
        discoveryService3.bindTopologyEventListener(listener3);

        instance3.heartbeatsAndCheckView();

        for(int i=0; i<4; i++) {
            Thread.sleep(500);
            instance1.heartbeatsAndCheckView();
            instance2.heartbeatsAndCheckView();
            instance3.heartbeatsAndCheckView();
        }

        Thread.sleep(joinerDelaySeconds * 1000);

        waitForEquals(listener3, 1, 1000);
        waitForEquals(listener1, 5, 1000);
        waitForEquals(listener2, 3, 1000);
    }

    @Test
    public void testSingleStart() throws Exception {
        // 1. start instance 1 normally
        OakVirtualInstanceBuilder builder1 =
                (OakVirtualInstanceBuilder) new OakVirtualInstanceBuilder()
                .setDebugName("instance1")
                .newRepository("/foo1/barrio/foo1/", true)
                .setConnectorPingInterval(999999)
                .setConnectorPingTimeout(999999);
        builder1.getConfig().setMinEventDelay(1);
        builder1.getConfig().setJoinerDelaySeconds(5);
        builder1.getConfig().setSyncTokenEnabled(true);
        builder1.getConfig().setSuppressPartiallyStartedInstance(true);
        // 1. start instance 1 normally -> call builder1.build()
        VirtualInstance instance1 = builder1.build();
        DummyListener listener1 = new DummyListener();
        OakDiscoveryService discoveryService1 = (OakDiscoveryService) instance1.getDiscoveryService();
        discoveryService1.bindTopologyEventListener(listener1);
        instance1.heartbeatsAndCheckView();

        Thread.sleep(1000);
        instance1.heartbeatsAndCheckView();
        Thread.sleep(1000);
        instance1.heartbeatsAndCheckView();
        assertEquals(1, listener1.countEvents());
    }

    @Test
    public void testSlowLeaderOnChange() throws Exception {
        // a leader reacting slow with syncToken update risks loosing
        // the leader flag. this test should ensure it's fine as long
        // as it is within minEventDelay
        OakVirtualInstanceBuilder builder1 =
                (OakVirtualInstanceBuilder) new OakVirtualInstanceBuilder()
                .setDebugName("instance1")
                .newRepository("/foo1/barrio/foo1/", true)
                .setConnectorPingInterval(999999)
                .setConnectorPingTimeout(999999);
        builder1.getConfig().setMinEventDelay(6);
        builder1.getConfig().setJoinerDelaySeconds(1);
        builder1.getConfig().setSyncTokenEnabled(true);
        builder1.getConfig().setSuppressPartiallyStartedInstance(true);

        // 1. start instance 1 normally -> call builder1.build()
        VirtualInstance instance1 = builder1.build();
        DummyListener listener1 = new DummyListener();
        OakDiscoveryService discoveryService1 = (OakDiscoveryService) instance1.getDiscoveryService();
        discoveryService1.bindTopologyEventListener(listener1);
        instance1.heartbeatsAndCheckView();
        assertEquals(0, listener1.countEvents());
        instance1.heartbeatsAndCheckView();
        waitForEquals(listener1, 1, 1000);
        assertEquals(1, listener1.countEvents());

        OakVirtualInstanceBuilder builder2 =
                (OakVirtualInstanceBuilder) new OakVirtualInstanceBuilder()
                .setDebugName("instance2")
                .useRepositoryOf(builder1)
                .setConnectorPingInterval(999999)
                .setConnectorPingTimeout(999999);
        builder2.getConfig().setMinEventDelay(6);
        builder2.getConfig().setJoinerDelaySeconds(1);
        builder2.getConfig().setSyncTokenEnabled(true);
        builder2.getConfig().setSuppressPartiallyStartedInstance(true);

        // 2. start instance 2 normally -> call builder2.build()
        VirtualInstance instance2 = builder2.build();
        DummyListener listener2 = new DummyListener();
        OakDiscoveryService discoveryService2 = (OakDiscoveryService) instance2.getDiscoveryService();
        discoveryService2.bindTopologyEventListener(listener2);

        // 3. now let instance 1 be slow, instance 2 fast, with updating syncToken
        // in any case, update the discovery-lite-view
        builder2.getSimulatedLeaseCollection().incSeqNum();
        builder2.updateLease();
        // but now don't update anything with instance1 as to not update the syncToken

        Thread.sleep(1000);
        instance2.heartbeatsAndCheckView();
        assertEquals(1, listener1.countEvents());
        assertEquals(0, listener2.countEvents());
        Thread.sleep(1000);
        instance2.heartbeatsAndCheckView();
        assertEquals(1, listener1.countEvents());
        assertEquals(0, listener2.countEvents());
        for(int i=0; i<8; i++) {
            Thread.sleep(1000);
            instance2.heartbeatsAndCheckView();
            assertEquals("i is " + i, 1, listener1.countEvents());
            assertEquals("i is " + i, 0, listener2.countEvents());
        }
        Thread.sleep(1000);
        instance1.heartbeatsAndCheckView();
        instance2.heartbeatsAndCheckView();
        Thread.sleep(1000);
        instance1.heartbeatsAndCheckView();
        instance2.heartbeatsAndCheckView();
        Thread.sleep(1000);
        instance1.heartbeatsAndCheckView();
        instance2.heartbeatsAndCheckView();
        for(int i=0; i<4; i++) {
            Thread.sleep(1000);
            instance1.heartbeatsAndCheckView();
            instance2.heartbeatsAndCheckView();
            assertEquals("i is " + i, 3, listener1.countEvents());
            assertEquals("i is " + i, 1, listener2.countEvents());
        }
    }

    @Test
    public void testFullPartialFullFullStart() throws Exception {
        // 1. start instance 1 normally
        // 2. start instance 2 only partially -> gets ignored by 1
        // 3. start instance 3 + 4 fully -> they should wait for (4/3 and 2)
        // 4. then start instance 2 fully -> 3 and 4 should get resolved
        OakVirtualInstanceBuilder builder1 =
                (OakVirtualInstanceBuilder) new OakVirtualInstanceBuilder()
                .setDebugName("instance1")
                .newRepository("/foo1/barrio/foo1/", true)
                .setConnectorPingInterval(999999)
                .setConnectorPingTimeout(999999);
        builder1.getConfig().setMinEventDelay(1);
        builder1.getConfig().setJoinerDelaySeconds(5);
        builder1.getConfig().setSyncTokenEnabled(true);
        builder1.getConfig().setSuppressPartiallyStartedInstance(true);

        // 1. start instance 1 normally -> call builder1.build()
        VirtualInstance instance1 = builder1.build();
        DummyListener listener1 = new DummyListener();
        OakDiscoveryService discoveryService1 = (OakDiscoveryService) instance1.getDiscoveryService();
        discoveryService1.bindTopologyEventListener(listener1);
        instance1.heartbeatsAndCheckView();

        Thread.sleep(1000);
        instance1.heartbeatsAndCheckView();
        Thread.sleep(1000);
        instance1.heartbeatsAndCheckView();
        assertEquals(1, listener1.countEvents());

        OakVirtualInstanceBuilder builder2 =
                (OakVirtualInstanceBuilder) new OakVirtualInstanceBuilder()
                .setDebugName("instance2")
                .useRepositoryOf(builder1)
                .setConnectorPingInterval(999999)
                .setConnectorPingTimeout(999999);
        builder2.getConfig().setJoinerDelaySeconds(5);
        builder2.getConfig().setSyncTokenEnabled(true);
        builder2.getConfig().setSuppressPartiallyStartedInstance(true);
        // 2. start instance 2 only partially -> do not call builder2.build() but just updateLease()
        builder2.updateLease();

        Thread.sleep(1000);

        OakVirtualInstanceBuilder builder3 =
                (OakVirtualInstanceBuilder) new OakVirtualInstanceBuilder()
                .setDebugName("instance3")
                .useRepositoryOf(builder1)
                .setConnectorPingInterval(999999)
                .setConnectorPingTimeout(999999);
        builder3.getConfig().setJoinerDelaySeconds(5);
        builder3.getConfig().setSyncTokenEnabled(true);
        builder3.getConfig().setSuppressPartiallyStartedInstance(true);
        // 3. start instance 3 + 4 fully -> builder3.build()
        VirtualInstance instance3 = builder3.build();
        DummyListener listener3 = new DummyListener();
        OakDiscoveryService discoveryService3 = (OakDiscoveryService) instance3.getDiscoveryService();
        discoveryService3.bindTopologyEventListener(listener3);

        OakVirtualInstanceBuilder builder4 =
                (OakVirtualInstanceBuilder) new OakVirtualInstanceBuilder()
                .setDebugName("instance4")
                .useRepositoryOf(builder1)
                .setConnectorPingInterval(999999)
                .setConnectorPingTimeout(999999);
        builder4.getConfig().setJoinerDelaySeconds(5);
        builder4.getConfig().setSyncTokenEnabled(true);
        builder4.getConfig().setSuppressPartiallyStartedInstance(true);
        // 3. start instance 3 + 4 fully -> builder4.build()
        VirtualInstance instance4 = builder4.build();
        DummyListener listener4 = new DummyListener();
        OakDiscoveryService discoveryService4 = (OakDiscoveryService) instance4.getDiscoveryService();
        discoveryService4.bindTopologyEventListener(listener4);

        instance1.heartbeatsAndCheckView();
        instance3.heartbeatsAndCheckView();
        instance4.heartbeatsAndCheckView();
        for(int i=0; i<7; i++) {
            Thread.sleep(1000);
            instance1.heartbeatsAndCheckView();
            instance3.heartbeatsAndCheckView();
            instance4.heartbeatsAndCheckView();
        }

        assertEquals(0, listener3.countEvents());
        assertEquals(0, listener4.countEvents());

        // 4. then start instance 2 fully -> 3 and 4 should get resolved
        VirtualInstance instance2 = builder2.build();

        instance1.heartbeatsAndCheckView();
        instance2.heartbeatsAndCheckView();
        instance3.heartbeatsAndCheckView();
        instance4.heartbeatsAndCheckView();
        assertEquals(0, listener3.countEvents());
        assertEquals(0, listener4.countEvents());
        for(int i=0; i<7; i++) {
            Thread.sleep(1000);
            instance1.heartbeatsAndCheckView();
            instance2.heartbeatsAndCheckView();
            instance3.heartbeatsAndCheckView();
            instance4.heartbeatsAndCheckView();
        }

        assertEquals(1, listener3.countEvents());
        assertEquals(1, listener4.countEvents());
    }

    @Test
    public void testUpdateDuringJoinDelay() throws Exception {
        OakVirtualInstanceBuilder builder1 =
                (OakVirtualInstanceBuilder) new OakVirtualInstanceBuilder()
                .setDebugName("instance1")
                .newRepository("/foo1/barrio/foo1/", true)
                .setConnectorPingInterval(999999)
                .setConnectorPingTimeout(999999);
        builder1.getConfig().setMinEventDelay(1);
        builder1.getConfig().setJoinerDelaySeconds(5);
        builder1.getConfig().setSyncTokenEnabled(true);
        builder1.getConfig().setSuppressPartiallyStartedInstance(true);
        // 1. start instance 1 normally -> call builder1.build()
        VirtualInstance instance1 = builder1.build();
        DummyListener listener1 = new DummyListener();
        OakDiscoveryService discoveryService1 = (OakDiscoveryService) instance1.getDiscoveryService();
        discoveryService1.bindTopologyEventListener(listener1);
        instance1.heartbeatsAndCheckView();

        Thread.sleep(1000);
        instance1.heartbeatsAndCheckView();
        Thread.sleep(1000);
        instance1.heartbeatsAndCheckView();
        assertEquals(1, listener1.countEvents());

        OakVirtualInstanceBuilder builder2 =
                (OakVirtualInstanceBuilder) new OakVirtualInstanceBuilder()
                .setDebugName("instance2")
                .useRepositoryOf(builder1)
                .setConnectorPingInterval(999999)
                .setConnectorPingTimeout(999999);
        builder2.getConfig().setJoinerDelaySeconds(6);
        builder2.getConfig().setSyncTokenEnabled(true);
        builder2.getConfig().setMinEventDelay(1);
        builder2.getConfig().setSuppressPartiallyStartedInstance(true);
        // 2. start instance 2 only partially -> do not call builder2.build() but just updateLease()
        builder2.updateLease();

        Thread.sleep(2000);
        assertEquals(1, listener1.countEvents());

        // joinDelay timer starts with building (plus 1sec for minEventDelay)
        VirtualInstance instance2 = builder2.build();
        DummyListener listener2 = new DummyListener();
        OakDiscoveryService discoveryService2 = (OakDiscoveryService) instance2.getDiscoveryService();
        discoveryService2.bindTopologyEventListener(listener2);

        Thread.sleep(1000);
        assertEquals(0, listener2.countEvents());

        instance1.heartbeatsAndCheckView();
        instance2.heartbeatsAndCheckView();
        Thread.sleep(1000);
        instance1.heartbeatsAndCheckView();
        instance2.heartbeatsAndCheckView();
        Thread.sleep(1000);
        instance1.heartbeatsAndCheckView();
        instance2.heartbeatsAndCheckView();

        assertEquals(3, listener1.countEvents());
        assertEquals(0, listener2.countEvents());

        OakVirtualInstanceBuilder builder3 =
                (OakVirtualInstanceBuilder) new OakVirtualInstanceBuilder()
                .setDebugName("instance3")
                .useRepositoryOf(builder1)
                .setConnectorPingInterval(999999)
                .setConnectorPingTimeout(999999);
        builder3.getConfig().setMinEventDelay(1);
        builder3.getConfig().setJoinerDelaySeconds(5);
        builder3.getConfig().setSyncTokenEnabled(true);
        builder3.getConfig().setSuppressPartiallyStartedInstance(true);
        VirtualInstance instance3 = builder3.build();
        DummyListener listener3 = new DummyListener();
        OakDiscoveryService discoveryService3 = (OakDiscoveryService) instance3.getDiscoveryService();
        discoveryService3.bindTopologyEventListener(listener3);

        // 3 started, now let 1 and 3 send heartbeats, 2 not
        instance1.heartbeatsAndCheckView();
        instance2.heartbeatsAndCheckView();
        instance3.heartbeatsAndCheckView();
        Thread.sleep(1000);
        instance1.heartbeatsAndCheckView();
        instance2.heartbeatsAndCheckView();
        instance3.heartbeatsAndCheckView();
        Thread.sleep(1000);
        instance1.heartbeatsAndCheckView();
        instance2.heartbeatsAndCheckView();
        instance3.heartbeatsAndCheckView();
        Thread.sleep(1000);
        // now it's 6sec, so the joinDelay of 2 should have timed out,
        // however, as we started instance 3 the view of 2 should now have changed
        instance1.heartbeatsAndCheckView();
        instance2.heartbeatsAndCheckView();
        instance3.heartbeatsAndCheckView();
        Thread.sleep(1000);
        assertEquals(5, listener1.countEvents());
        assertEquals(1, listener2.countEvents());

        instance1.heartbeatsAndCheckView();
        instance2.heartbeatsAndCheckView();
        instance3.heartbeatsAndCheckView();
        Thread.sleep(1000);
        instance1.heartbeatsAndCheckView();
        instance2.heartbeatsAndCheckView();
        instance3.heartbeatsAndCheckView();
        Thread.sleep(1000);
        instance1.heartbeatsAndCheckView();
        instance2.heartbeatsAndCheckView();
        instance3.heartbeatsAndCheckView();

        // check all the listeners
        waitForEquals(listener1, 5, 1000);
        waitForEquals(listener2, 1, 1000);
        waitForEquals(listener3, 1, 1000);
    }

    /**
     * This test requires the minEventHandlerDelay to be configured properly, ie reasonably
     * high to delay the time between TOPOLOGY_CHANGING and TOPOLOGY_CHANGED
     * (as a replacement of previously not suppressing mismatching synctokens)
     */
    @Test
    public void reuseClusterIdTest() throws Exception {
        logger.info("reuseClusterIdTest : START - creating instance1.");
        OakVirtualInstanceBuilder builder1 = createInstanceWithNewRepo("instance", "/foo1/barrio/foo1/", 1, 5);
        // 1. start instance 1 normally -> call builder1.build()
        VirtualInstance instance1 = builder1.build();
        DummyListener listener1 = createDummyListener(instance1);

        for(int i = 0; i < 3; i++) {
            instance1.heartbeatsAndCheckView();
            Thread.sleep(1000);
        }
        assertEquals(1, listener1.countEvents());

        // 2. start instance 2 partially
        logger.info("reuseClusterIdTest : creating instance2.");
        OakVirtualInstanceBuilder builder2 = createInstanceUsingExistingRepo(builder1, "instance", 1, 5);
        // 2. start instance 2 only partially -> do not call builder2.build() but just updateLease()
        builder2.updateLease();

        for(int i = 0; i < 3; i++) {
            instance1.heartbeatsAndCheckView();
            Thread.sleep(1000);
        }
        assertEquals(1, listener1.countEvents());

        // finish instance2 startup
        VirtualInstance instance2 = builder2.build();
        DummyListener listener2 = createDummyListener(instance2);

        for(int i = 0; i < 3; i++) {
            instance1.heartbeatsAndCheckView();
            instance2.heartbeatsAndCheckView();
            Thread.sleep(1000);
        }
        assertEquals(0, listener2.countEvents());
        assertEquals(3, listener1.countEvents());

        for(int i = 0; i < 3; i++) {
            instance1.heartbeatsAndCheckView();
            instance2.heartbeatsAndCheckView();
            Thread.sleep(1000);
        }
        assertEquals(1, listener2.countEvents());
        assertEquals(3, listener1.countEvents());

        // stop instance2
        logger.info("reuseClusterIdTest : STOP 2");
        instance2.stop();

        for(int i = 0; i < 2; i++) {
            instance1.heartbeatsAndCheckView();
            Thread.sleep(1000);
        }
        assertEquals(5, listener1.countEvents());

        // re-create instance2, with same clusterNodeId/slingId
        logger.info("reuseClusterIdTest : creating instance2b.");
        OakVirtualInstanceBuilder builder2b = createInstanceUsingExistingRepo(builder1,
                "instance2b", 1, 5);
        builder2b.setSlingId(builder2.getSlingId());
        builder2b.getLease().setClusterNodeIdHint(2);
        builder2b.getSimulatedLeaseCollection().incSeqNum();
        // 2. start instance 2 only partially -> do not call builder2.build() but just updateLease()
        builder2b.updateLease();

        for(int i = 0; i < 2; i++) {
            instance1.heartbeatsAndCheckView();
            Thread.sleep(1000);
        }
        instance1.heartbeatsAndCheckView();
        assertEquals(5 /*not yet CHANGING */, listener1.countEvents());

        // finish instance2 startup
        VirtualInstance instance2b = builder2b.build();
        DummyListener listener2b = createDummyListener(instance2b);

        for(int i = 0; i < 3; i++) {
            instance1.heartbeatsAndCheckView();
            instance2b.heartbeatsAndCheckView();
            Thread.sleep(1000);
        }
        instance1.heartbeatsAndCheckView();
        instance2b.heartbeatsAndCheckView();
        assertEquals(0, listener2b.countEvents());
        assertEquals(7, listener1.countEvents());

        for(int i = 0; i < 4; i++) {
            Thread.sleep(1000);
            instance1.heartbeatsAndCheckView();
            instance2b.heartbeatsAndCheckView();
        }
        waitForEquals(listener2b, 1, 2000);
        waitForEquals(listener1, 7, 2000); // had a flaky failure so giving it 2 sec
    }

    @Test
    public void testTopologyChangeWithOneSlowMember() throws Exception {
        logger.info("testTopologyChangeWithOneSlowMember : START - creating instance1.");
        OakVirtualInstanceBuilder builder1 = createInstanceWithNewRepo("instance1",
                "/foo1/barrio/foo1/", 5, 1);
        // 1. start instance 1 normally -> call builder1.build()
        VirtualInstance instance1 = builder1.build();
        DummyListener listener1 = createDummyListener(instance1);

        // instance2 normally
        logger.info("testTopologyChangeWithOneSlowMember : creating instance2.");
        OakVirtualInstanceBuilder builder2 = createInstanceUsingExistingRepo(builder1,
                "instance2", 1, 1);
        // 2. start instance 2 normally -> call builder2.build()
        VirtualInstance instance2 = builder2.build();
        DummyListener listener2 = createDummyListener(instance2);

        // a bunch of heartbeats
        for(int i=0; i<7; i++) {
            instance1.heartbeatsAndCheckView();
            instance2.heartbeatsAndCheckView();
            Thread.sleep(1000);
        }

        logger.info("testTopologyChangeWithOneSlowMember : check normal situation established");
        assertTrue("actual : " + listener1.countEvents(), listener1.countEvents() == 1 || listener1.countEvents() == 3);
        assertTrue("actual : " + listener2.countEvents(), listener2.countEvents() == 1 || listener2.countEvents() == 3);

        listener1.clearEvents();
        listener2.clearEvents();

        // trigger a lightning-fast change in the discovery-lite
        builder1.getLease().incSeqNum();
        // inc seqnum by 2, then it is actually valid (an instance joining/leaving quickly)
        builder1.getLease().incSeqNum();

        // now let instance2 behave slow : it still has a valid lease but doesn't do the syncToken quickly
        for(int i=0; i<4; i++) {
            instance1.heartbeatsAndCheckView();
            Thread.sleep(1000);
        }
        logger.info("testTopologyChangeWithOneSlowMember : check listener1 having gotten a TOPOLOGY_CHANGING but listener2 not yet");
        assertEquals(1, listener1.countEvents()); // a TOPOLOGY_CHANGING only
        assertEquals(TopologyEvent.Type.TOPOLOGY_CHANGING, listener1.getEvents().get(0).getType());
        assertEquals(0, listener2.countEvents());
    }

    private OakVirtualInstanceBuilder createInstanceUsingExistingRepo(
            OakVirtualInstanceBuilder builder1, String debugName, int minEventDelay, int joinerDelay) throws Exception {
        OakVirtualInstanceBuilder builder2 =
                (OakVirtualInstanceBuilder) new OakVirtualInstanceBuilder()
                .setDebugName(debugName)
                .useRepositoryOf(builder1)
                .setConnectorPingInterval(999999)
                .setConnectorPingTimeout(999999);
        builder2.getConfig().setMinEventDelay(minEventDelay);
        builder2.getConfig().setJoinerDelaySeconds(joinerDelay);
        builder2.getConfig().setSyncTokenEnabled(true);
        builder2.getConfig().setSuppressPartiallyStartedInstance(true);
        return builder2;
    }

    private DummyListener createDummyListener(VirtualInstance virtualInstance) {
        final DummyListener newListener = new DummyListener();
        final OakDiscoveryService ds = (OakDiscoveryService) virtualInstance.getDiscoveryService();
        ds.bindTopologyEventListener(newListener);
        return newListener;
    }

    private OakVirtualInstanceBuilder createInstanceWithNewRepo(String debugName,
            String discoveryPath, int minEventDelay, long joinerDelay) throws Exception {
        final OakVirtualInstanceBuilder builder = (OakVirtualInstanceBuilder) new OakVirtualInstanceBuilder()
                .setDebugName(debugName).newRepository(discoveryPath, true)
                .setConnectorPingInterval(999999).setConnectorPingTimeout(999999);
        builder.getConfig().setMinEventDelay(minEventDelay);
        builder.setMinEventDelay(minEventDelay);
        builder.getConfig().setJoinerDelaySeconds(joinerDelay);
        builder.getConfig().setSyncTokenEnabled(true);
        builder.getConfig().setSuppressPartiallyStartedInstance(true);
        return builder;
    }
}