<commit_before>/* This file is part of VoltDB.
 * Copyright (C) 2008-2017 VoltDB Inc.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as
 * published by the Free Software Foundation, either version 3 of the
 * License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with VoltDB.  If not, see <http://www.gnu.org/licenses/>.
 */

package org.voltdb.importer;

import static com.google_voltpatches.common.base.Preconditions.checkArgument;
import static com.google_voltpatches.common.base.Preconditions.checkNotNull;
import static com.google_voltpatches.common.base.Preconditions.checkState;
import static com.google_voltpatches.common.base.Predicates.equalTo;
import static com.google_voltpatches.common.base.Predicates.isNull;
import static com.google_voltpatches.common.base.Predicates.not;
import static com.google_voltpatches.common.base.Predicates.or;
import static org.voltcore.zk.ZKUtil.joinZKPath;

import java.io.File;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicStampedReference;

import org.apache.zookeeper_voltpatches.AsyncCallback;
import org.apache.zookeeper_voltpatches.AsyncCallback.Children2Callback;
import org.apache.zookeeper_voltpatches.AsyncCallback.DataCallback;
import org.apache.zookeeper_voltpatches.AsyncCallback.StatCallback;
import org.apache.zookeeper_voltpatches.AsyncCallback.StringCallback;
import org.apache.zookeeper_voltpatches.AsyncCallback.VoidCallback;
import org.apache.zookeeper_voltpatches.CreateMode;
import org.apache.zookeeper_voltpatches.KeeperException;
import org.apache.zookeeper_voltpatches.KeeperException.Code;
import org.apache.zookeeper_voltpatches.KeeperException.NodeExistsException;
import org.apache.zookeeper_voltpatches.WatchedEvent;
import org.apache.zookeeper_voltpatches.Watcher;
import org.apache.zookeeper_voltpatches.Watcher.Event.EventType;
import org.apache.zookeeper_voltpatches.Watcher.Event.KeeperState;
import org.apache.zookeeper_voltpatches.ZooDefs;
import org.apache.zookeeper_voltpatches.ZooKeeper;
import org.apache.zookeeper_voltpatches.data.Stat;
import org.json_voltpatches.JSONArray;
import org.json_voltpatches.JSONException;
import org.json_voltpatches.JSONStringer;
import org.voltcore.logging.VoltLogger;
import org.voltcore.utils.CoreUtils;
import org.voltcore.zk.ZKUtil;
import org.voltdb.OperationMode;
import org.voltdb.VoltDB;
import org.voltdb.VoltZK;
import com.google_voltpatches.common.base.Function;
import com.google_voltpatches.common.base.Optional;
import com.google_voltpatches.common.base.Preconditions;
import com.google_voltpatches.common.base.Predicate;
import com.google_voltpatches.common.collect.FluentIterable;
import com.google_voltpatches.common.collect.ImmutableSortedMap;
import com.google_voltpatches.common.collect.ImmutableSortedSet;
import com.google_voltpatches.common.collect.Maps;
import com.google_voltpatches.common.collect.Sets;
import com.google_voltpatches.common.collect.TreeMultimap;
import com.google_voltpatches.common.eventbus.AsyncEventBus;
import com.google_voltpatches.common.eventbus.DeadEvent;
import com.google_voltpatches.common.eventbus.EventBus;
import com.google_voltpatches.common.eventbus.Subscribe;
import com.google_voltpatches.common.eventbus.SubscriberExceptionContext;
import com.google_voltpatches.common.eventbus.SubscriberExceptionHandler;
import com.google_voltpatches.common.util.concurrent.SettableFuture;

/**
 *  An importer channel distributer that uses zookeeper to coordinate how importer channels are
 *  distributed among VoltDB cluster nodes. A proposal is merged against a master channel lists,
 *  stored in the /import/master zookeeper node. The elected distributer leader distributes the
 *  merge differences among all available nodes, by writing each node's assigned list of
 *  channels in their respective /import/host/[host-name] nodes. When a node leaves the mesh
 *  its assigned channels are redistributed among the surviving nodes
 */
public class ChannelDistributer implements ChannelChangeCallback {

    private final static VoltLogger LOG = new VoltLogger("IMPORT");

    /** root for all importer nodes */
    static final String IMPORT_DN = "/import";
    /** parent node for all {@link CreateMode#EPHEMERAL ephemeral} host nodes */
    static final String HOST_DN = joinZKPath(IMPORT_DN, "host");
    /** parent directory for leader candidates and holder of the channels master list */
    static final String MASTER_DN = joinZKPath(IMPORT_DN, "master");
    /** leader candidate node prefix for {@link CreateMode#EPHEMERAL_SEQUENTIAL} nodes */
    static final String CANDIDATE_PN = joinZKPath(MASTER_DN, "candidate_");

    static final byte[] EMPTY_ARRAY = "[]".getBytes(StandardCharsets.UTF_8);

    static void mkdirs(ZooKeeper zk, String zkNode, byte[] content) {
        try {
            ZKUtil.asyncMkdirs(zk, zkNode, content).get();
        } catch (NodeExistsException itIsOk) {
        } catch (InterruptedException | KeeperException e) {
            String msg = "Unable to create zk directory: " + zkNode;
            LOG.error(msg, e);
            throw new DistributerException(msg, e);
        }
    }

    /**
     * Boiler plate method to log an error message and wrap, and return a {@link DistributerException}
     * around the message and cause
     *
     * @param cause fault origin {@link Throwable}
     * @param format a {@link String#format(String, Object...) compliant format string
     * @param args formatter arguments
     * @return a {@link DistributerException}
     */
    static DistributerException loggedDistributerException(Throwable cause, String format, Object...args) {
        Optional<DistributerException> causeFor = DistributerException.isCauseFor(cause);
        if (causeFor.isPresent()) {
            return causeFor.get();
        }
        String msg = String.format(format, args);
        if (cause != null) {
            LOG.error(msg, cause);
            return new DistributerException(msg, cause);
        } else {
            LOG.error(msg);
            return new DistributerException(msg);
        }
    }

    /**
     * Boiler plate method that checks a zookeeper callback @{link {@link KeeperException.Code},
     * converts it to a {@link DistributerException} and if it does not indicate success,
     *
     * @param code a {@link KeeperException.Code callback code}
     * @param format {@link String#format(String, Object...) compliant format string
     * @param args formatter arguments
     * @return an {@link Optional} that may contain a {@link DistributerException}
     */
    static Optional<DistributerException> checkCode(Code code, String format, Object...args) {
        if (code != Code.OK) {
            KeeperException kex = KeeperException.create(code);
            return Optional.of(loggedDistributerException(kex, format, args));
        } else {
            return Optional.absent();
        }
    }

    /**
     * Boiler plate method that acquires, and releases a {@link Semaphore}
     * @param lock a {@link Semaphore}
     */
    static void acquireAndRelease(Semaphore lock) {
        try {
            lock.acquire();
            lock.release();
        } catch (InterruptedException ex) {
            throw loggedDistributerException(ex, "interruped while waiting for a semaphare");
        }
    }

    /**
     * Reads the JSON document contained in the byte array data, and
     * converts it to a {@link NavigableSet<ChannelSpec> set of channel specs}
     *
     * @param data zookeeper node data content
     * @return a {@link NavigableSet<ChannelSpec> set of channel specs}
     * @throws JSONException on JSON parse failures
     * @throws IllegalArgumentException on encoded channel spec parse failures
     */
    static NavigableSet<ChannelSpec> asChannelSet(byte[] data)
            throws JSONException, IllegalArgumentException {
        ImmutableSortedSet.Builder<ChannelSpec> sbld = ImmutableSortedSet.naturalOrder();
        JSONArray ja = new JSONArray(new String(data, StandardCharsets.UTF_8));
        for (int i=0; i< ja.length(); ++i) {
            sbld.add(new ChannelSpec(ja.getString(i)));
        }
        return  sbld.build();
    }

    /**
     * Converts the given a {@link NavigableSet<ChannelSpec> set of channel specs}
     * into a byte array with the content of a JSON document
     *
     * @param specs a a {@link NavigableSet<ChannelSpec> set of channel specs}
     * @return a byte array
     * @throws JSONException on JSON building failures
     * @throws IllegalArgumentException on channel spec encoding failures
     */
    static byte [] asHostData(NavigableSet<ChannelSpec> specs)
            throws JSONException, IllegalArgumentException {
        JSONStringer js = new JSONStringer();
        js.array();
        for (ChannelSpec spec: specs) {
            js.value(spec.asJSONValue());
        }
        js.endArray();
        return js.toString().getBytes(StandardCharsets.UTF_8);
    }

    /**
     * Tracing utility method useful for debugging
     *
     * @param o and object
     * @return string with information on the given object
     */
    static String id(Object o) {
        if (o == null) return "(null)";
        Thread t = Thread.currentThread();
        StringBuilder sb = new StringBuilder(128);
        sb.append("(T[").append(t.getName()).append("]@");
        sb.append(Long.toString(t.getId(), Character.MAX_RADIX));
        sb.append(":O[").append(o.getClass().getSimpleName());
        sb.append("]@");
        sb.append(Long.toString(System.identityHashCode(o),Character.MAX_RADIX));
        sb.append(")");
        return sb.toString();
    }

    final static SubscriberExceptionHandler eventBusFaultHandler = new SubscriberExceptionHandler() {
        @Override
        public void handleException(Throwable exception, SubscriberExceptionContext context) {
            loggedDistributerException(
                    exception,
                    "fault during callback dispatch for event %s",
                    context.getEvent()
                    );
        }
    };

    private final ExecutorService m_es;
    private final AtomicBoolean m_done = new AtomicBoolean(false);
    private final ZooKeeper m_zk;
    private final String m_hostId;
    private final String m_candidate;
    private final Deque<ImporterChannelAssignment> m_undispatched;
    private final EventBus m_eb;
    private final ExecutorService m_buses;

    volatile boolean m_isLeader = false;
    final SpecsRef m_specs = new SpecsRef();
    final HostsRef m_hosts = new HostsRef();
    final ChannelsRef m_channels = new ChannelsRef();
    final CallbacksRef m_callbacks = new CallbacksRef();
    final UnregisteredRef m_unregistered = new UnregisteredRef();
    final AtomicStampedReference<OperationMode> m_mode;

    /**
     * Initialize a distributer within importer channel distribution mesh by performing the
     * following actions:
     * <ul>
     * <li>registers a leader candidate</li>
     * <li>starts watchers on the channel master list, on the directory holding host nodes,
     * and the directory used for leader elections</li>
     * <li>create election candidate node</li>
     * <li>create its own host node</li>
     * </ul>
     * @param zk
     * @param hostId
     * @param queue
     */
    public ChannelDistributer(ZooKeeper zk, String hostId) {
        Preconditions.checkArgument(
                hostId != null && !hostId.trim().isEmpty(),
                "hostId is null or empty"
                );
        m_hostId = hostId;
        m_zk = Preconditions.checkNotNull(zk, "zookeeper is null");
        m_es = CoreUtils.getCachedSingleThreadExecutor("Import Channel Distributer for Host " + hostId, 15000);
        m_buses = CoreUtils.getCachedSingleThreadExecutor(
                "Import Channel Distributer Event Bus Dispatcher for Host " + hostId, 15000
                );
        m_eb = new AsyncEventBus(m_buses, eventBusFaultHandler);
        m_eb.register(this);
        m_mode = new AtomicStampedReference<>(OperationMode.RUNNING, 0);
        m_undispatched = new LinkedList<>();

        // Prime directory structure if needed
        OperationMode startMode = VoltDB.instance().getStartMode();
        assert startMode == OperationMode.RUNNING || startMode == OperationMode.PAUSED;
        mkdirs(zk, VoltZK.operationMode, startMode.getBytes());
        mkdirs(zk, HOST_DN, EMPTY_ARRAY);
        mkdirs(zk, MASTER_DN, EMPTY_ARRAY);

        GetOperationMode opMode = new GetOperationMode(VoltZK.operationMode);
        CreateNode createHostNode = new CreateNode(
                joinZKPath(HOST_DN, hostId),
                EMPTY_ARRAY, CreateMode.EPHEMERAL
                );
        MonitorHostNodes monitor = new MonitorHostNodes(HOST_DN);
        CreateNode electionCandidate = new CreateNode(
                CANDIDATE_PN,
                EMPTY_ARRAY, CreateMode.EPHEMERAL_SEQUENTIAL
                );
        ElectLeader elector = new ElectLeader(MASTER_DN, electionCandidate);

        createHostNode.getNode();
        elector.elect();

        m_candidate = electionCandidate.getNode();
        opMode.getMode();
        monitor.getChildren();
        // monitor the master list
        new GetChannels(MASTER_DN).getChannels();
    }

    public String getHostId() {
        return m_hostId;
    }

    public VersionedOperationMode getOperationMode() {
        return new VersionedOperationMode(m_mode);
    }

    /**
     * Register channels for the given importer. If they match to what is already registered
     * then nothing is done. Before registering channels, you need to register a callback
     * handler for channel assignments {@link #registerCallback(String, ChannelChangeCallback)}
     *
     * @param importer importer designation
     * @param uris list of channel URIs
     */
    public void registerChannels(String importer, Set<URI> uris) {
        NavigableSet<String> registered = m_callbacks.getReference().navigableKeySet();
        Preconditions.checkArgument(
                importer != null && !importer.trim().isEmpty(),
                "importer is null or empty"
                );
        Preconditions.checkArgument(uris != null, "uris set is null");
        Preconditions.checkArgument(
                !FluentIterable.from(uris).anyMatch(isNull()),
                "uris set %s contains null elements", uris
                );
        if (!registered.contains(importer)) {
            if (uris.isEmpty()) {
                // ImporterLifeCycleManager.stop() calls registerChannels() is called with an empty set of URIs.
                // If the importer never finished starting, we hit this condition.
                // This log message is used by the TestImporterStopAfterIncompleteStart JUnit.
                LOG.info("Skipping channel un-registration for " + importer + " since it did not finish initialization");
                return;
            } else {
                throw new IllegalStateException("no callbacks registered for " + importer
                        + " - unable to register channels " + Arrays.toString(uris.toArray()));
            }
        }

        Predicate<ChannelSpec> forImporter = ChannelSpec.importerIs(importer);
        Function<URI,ChannelSpec> asSpec = ChannelSpec.fromUri(importer);

        // convert method parameters to a set of ChannelSpecs
        NavigableSet<ChannelSpec> proposed = ImmutableSortedSet.copyOf(
                FluentIterable.from(uris).transform(asSpec)
                );

        LOG.info("(" + m_hostId + ") proposing channels " + proposed);

        int [] stamp = new int[]{0};

        ImmutableSortedSet.Builder<ChannelSpec> sbldr = null;
        NavigableSet<ChannelSpec> prev = null;
        SetData setter = null;
        NavigableSet<ChannelSpec> masterList = null;
        // retry writes when merging with stale data
        do {
            prev = m_channels.get(stamp);

            NavigableSet<ChannelSpec> current  = Sets.filter(prev, forImporter);
            if (current.equals(proposed)) {
                return;
            }
            sbldr = ImmutableSortedSet.naturalOrder();
            sbldr.addAll(Sets.filter(prev, not(forImporter)));
            sbldr.addAll(proposed);

            byte [] data = null;
            try {
                masterList = sbldr.build();
                data = asHostData(masterList);
            } catch (JSONException|IllegalArgumentException e) {
                throw loggedDistributerException(e, "failed to serialize the registration as json");
            }

            setter = new SetData(MASTER_DN, stamp[0], data);
        } while (setter.getCallbackCode() == Code.BADVERSION);

        //synch master channel list after channel registrations.
        int [] sstamp = new int[]{0};
        prev = m_channels.get(sstamp);
        m_channels.compareAndSet(prev, masterList, sstamp[0], stamp[0]);

        setter.getStat();
    }

    /**
     * Registers a (@link ChannelChangeCallback} for the given importer.
     * @param importer
     * @param callback a (@link ChannelChangeCallback}
     */
    public void registerCallback(String importer, ChannelChangeCallback callback) {
        Preconditions.checkArgument(
                importer != null && !importer.trim().isEmpty(),
                "importer is null or empty"
                );
        Preconditions.checkArgument(
                !m_unregistered.getReference().contains(importer),
                "cannot re-register importer %s as it was already unregistered",
                importer
                );
        callback = checkNotNull(callback, "callback is null");

        if (m_done.get()) return;

        int [] stamp = new int[]{0};
        NavigableMap<String,ChannelChangeCallback> prev = null;
        NavigableMap<String,ChannelChangeCallback> next = null;
        ImmutableSortedMap.Builder<String,ChannelChangeCallback> mbldr = null;

        synchronized (m_undispatched) {
            do {
                prev = m_callbacks.get(stamp);
                mbldr = ImmutableSortedMap.naturalOrder();
                mbldr.putAll(Maps.filterKeys(prev, not(equalTo(importer))));<commit_msg>ENG-13473 allow unrergistered importers to be re-registered<commit_after>/* This file is part of VoltDB.
 * Copyright (C) 2008-2017 VoltDB Inc.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as
 * published by the Free Software Foundation, either version 3 of the
 * License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with VoltDB.  If not, see <http://www.gnu.org/licenses/>.
 */

package org.voltdb.importer;

import static com.google_voltpatches.common.base.Preconditions.checkArgument;
import static com.google_voltpatches.common.base.Preconditions.checkNotNull;
import static com.google_voltpatches.common.base.Preconditions.checkState;
import static com.google_voltpatches.common.base.Predicates.equalTo;
import static com.google_voltpatches.common.base.Predicates.isNull;
import static com.google_voltpatches.common.base.Predicates.not;
import static com.google_voltpatches.common.base.Predicates.or;
import static org.voltcore.zk.ZKUtil.joinZKPath;

import java.io.File;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicStampedReference;

import org.apache.zookeeper_voltpatches.AsyncCallback;
import org.apache.zookeeper_voltpatches.AsyncCallback.Children2Callback;
import org.apache.zookeeper_voltpatches.AsyncCallback.DataCallback;
import org.apache.zookeeper_voltpatches.AsyncCallback.StatCallback;
import org.apache.zookeeper_voltpatches.AsyncCallback.StringCallback;
import org.apache.zookeeper_voltpatches.AsyncCallback.VoidCallback;
import org.apache.zookeeper_voltpatches.CreateMode;
import org.apache.zookeeper_voltpatches.KeeperException;
import org.apache.zookeeper_voltpatches.KeeperException.Code;
import org.apache.zookeeper_voltpatches.KeeperException.NodeExistsException;
import org.apache.zookeeper_voltpatches.WatchedEvent;
import org.apache.zookeeper_voltpatches.Watcher;
import org.apache.zookeeper_voltpatches.Watcher.Event.EventType;
import org.apache.zookeeper_voltpatches.Watcher.Event.KeeperState;
import org.apache.zookeeper_voltpatches.ZooDefs;
import org.apache.zookeeper_voltpatches.ZooKeeper;
import org.apache.zookeeper_voltpatches.data.Stat;
import org.json_voltpatches.JSONArray;
import org.json_voltpatches.JSONException;
import org.json_voltpatches.JSONStringer;
import org.voltcore.logging.VoltLogger;
import org.voltcore.utils.CoreUtils;
import org.voltcore.zk.ZKUtil;
import org.voltdb.OperationMode;
import org.voltdb.VoltDB;
import org.voltdb.VoltZK;
import com.google_voltpatches.common.base.Function;
import com.google_voltpatches.common.base.Optional;
import com.google_voltpatches.common.base.Preconditions;
import com.google_voltpatches.common.base.Predicate;
import com.google_voltpatches.common.collect.FluentIterable;
import com.google_voltpatches.common.collect.ImmutableSortedMap;
import com.google_voltpatches.common.collect.ImmutableSortedSet;
import com.google_voltpatches.common.collect.Maps;
import com.google_voltpatches.common.collect.Sets;
import com.google_voltpatches.common.collect.TreeMultimap;
import com.google_voltpatches.common.eventbus.AsyncEventBus;
import com.google_voltpatches.common.eventbus.DeadEvent;
import com.google_voltpatches.common.eventbus.EventBus;
import com.google_voltpatches.common.eventbus.Subscribe;
import com.google_voltpatches.common.eventbus.SubscriberExceptionContext;
import com.google_voltpatches.common.eventbus.SubscriberExceptionHandler;
import com.google_voltpatches.common.util.concurrent.SettableFuture;

/**
 *  An importer channel distributer that uses zookeeper to coordinate how importer channels are
 *  distributed among VoltDB cluster nodes. A proposal is merged against a master channel lists,
 *  stored in the /import/master zookeeper node. The elected distributer leader distributes the
 *  merge differences among all available nodes, by writing each node's assigned list of
 *  channels in their respective /import/host/[host-name] nodes. When a node leaves the mesh
 *  its assigned channels are redistributed among the surviving nodes
 */
public class ChannelDistributer implements ChannelChangeCallback {

    private final static VoltLogger LOG = new VoltLogger("IMPORT");

    /** root for all importer nodes */
    static final String IMPORT_DN = "/import";
    /** parent node for all {@link CreateMode#EPHEMERAL ephemeral} host nodes */
    static final String HOST_DN = joinZKPath(IMPORT_DN, "host");
    /** parent directory for leader candidates and holder of the channels master list */
    static final String MASTER_DN = joinZKPath(IMPORT_DN, "master");
    /** leader candidate node prefix for {@link CreateMode#EPHEMERAL_SEQUENTIAL} nodes */
    static final String CANDIDATE_PN = joinZKPath(MASTER_DN, "candidate_");

    static final byte[] EMPTY_ARRAY = "[]".getBytes(StandardCharsets.UTF_8);

    static void mkdirs(ZooKeeper zk, String zkNode, byte[] content) {
        try {
            ZKUtil.asyncMkdirs(zk, zkNode, content).get();
        } catch (NodeExistsException itIsOk) {
        } catch (InterruptedException | KeeperException e) {
            String msg = "Unable to create zk directory: " + zkNode;
            LOG.error(msg, e);
            throw new DistributerException(msg, e);
        }
    }

    /**
     * Boiler plate method to log an error message and wrap, and return a {@link DistributerException}
     * around the message and cause
     *
     * @param cause fault origin {@link Throwable}
     * @param format a {@link String#format(String, Object...) compliant format string
     * @param args formatter arguments
     * @return a {@link DistributerException}
     */
    static DistributerException loggedDistributerException(Throwable cause, String format, Object...args) {
        Optional<DistributerException> causeFor = DistributerException.isCauseFor(cause);
        if (causeFor.isPresent()) {
            return causeFor.get();
        }
        String msg = String.format(format, args);
        if (cause != null) {
            LOG.error(msg, cause);
            return new DistributerException(msg, cause);
        } else {
            LOG.error(msg);
            return new DistributerException(msg);
        }
    }

    /**
     * Boiler plate method that checks a zookeeper callback @{link {@link KeeperException.Code},
     * converts it to a {@link DistributerException} and if it does not indicate success,
     *
     * @param code a {@link KeeperException.Code callback code}
     * @param format {@link String#format(String, Object...) compliant format string
     * @param args formatter arguments
     * @return an {@link Optional} that may contain a {@link DistributerException}
     */
    static Optional<DistributerException> checkCode(Code code, String format, Object...args) {
        if (code != Code.OK) {
            KeeperException kex = KeeperException.create(code);
            return Optional.of(loggedDistributerException(kex, format, args));
        } else {
            return Optional.absent();
        }
    }

    /**
     * Boiler plate method that acquires, and releases a {@link Semaphore}
     * @param lock a {@link Semaphore}
     */
    static void acquireAndRelease(Semaphore lock) {
        try {
            lock.acquire();
            lock.release();
        } catch (InterruptedException ex) {
            throw loggedDistributerException(ex, "interruped while waiting for a semaphare");
        }
    }

    /**
     * Reads the JSON document contained in the byte array data, and
     * converts it to a {@link NavigableSet<ChannelSpec> set of channel specs}
     *
     * @param data zookeeper node data content
     * @return a {@link NavigableSet<ChannelSpec> set of channel specs}
     * @throws JSONException on JSON parse failures
     * @throws IllegalArgumentException on encoded channel spec parse failures
     */
    static NavigableSet<ChannelSpec> asChannelSet(byte[] data)
            throws JSONException, IllegalArgumentException {
        ImmutableSortedSet.Builder<ChannelSpec> sbld = ImmutableSortedSet.naturalOrder();
        JSONArray ja = new JSONArray(new String(data, StandardCharsets.UTF_8));
        for (int i=0; i< ja.length(); ++i) {
            sbld.add(new ChannelSpec(ja.getString(i)));
        }
        return  sbld.build();
    }

    /**
     * Converts the given a {@link NavigableSet<ChannelSpec> set of channel specs}
     * into a byte array with the content of a JSON document
     *
     * @param specs a a {@link NavigableSet<ChannelSpec> set of channel specs}
     * @return a byte array
     * @throws JSONException on JSON building failures
     * @throws IllegalArgumentException on channel spec encoding failures
     */
    static byte [] asHostData(NavigableSet<ChannelSpec> specs)
            throws JSONException, IllegalArgumentException {
        JSONStringer js = new JSONStringer();
        js.array();
        for (ChannelSpec spec: specs) {
            js.value(spec.asJSONValue());
        }
        js.endArray();
        return js.toString().getBytes(StandardCharsets.UTF_8);
    }

    /**
     * Tracing utility method useful for debugging
     *
     * @param o and object
     * @return string with information on the given object
     */
    static String id(Object o) {
        if (o == null) return "(null)";
        Thread t = Thread.currentThread();
        StringBuilder sb = new StringBuilder(128);
        sb.append("(T[").append(t.getName()).append("]@");
        sb.append(Long.toString(t.getId(), Character.MAX_RADIX));
        sb.append(":O[").append(o.getClass().getSimpleName());
        sb.append("]@");
        sb.append(Long.toString(System.identityHashCode(o),Character.MAX_RADIX));
        sb.append(")");
        return sb.toString();
    }

    final static SubscriberExceptionHandler eventBusFaultHandler = new SubscriberExceptionHandler() {
        @Override
        public void handleException(Throwable exception, SubscriberExceptionContext context) {
            loggedDistributerException(
                    exception,
                    "fault during callback dispatch for event %s",
                    context.getEvent()
                    );
        }
    };

    private final ExecutorService m_es;
    private final AtomicBoolean m_done = new AtomicBoolean(false);
    private final ZooKeeper m_zk;
    private final String m_hostId;
    private final String m_candidate;
    private final Deque<ImporterChannelAssignment> m_undispatched;
    private final EventBus m_eb;
    private final ExecutorService m_buses;

    volatile boolean m_isLeader = false;
    final SpecsRef m_specs = new SpecsRef();
    final HostsRef m_hosts = new HostsRef();
    final ChannelsRef m_channels = new ChannelsRef();
    final CallbacksRef m_callbacks = new CallbacksRef();
    final UnregisteredRef m_unregistered = new UnregisteredRef();
    final AtomicStampedReference<OperationMode> m_mode;

    /**
     * Initialize a distributer within importer channel distribution mesh by performing the
     * following actions:
     * <ul>
     * <li>registers a leader candidate</li>
     * <li>starts watchers on the channel master list, on the directory holding host nodes,
     * and the directory used for leader elections</li>
     * <li>create election candidate node</li>
     * <li>create its own host node</li>
     * </ul>
     * @param zk
     * @param hostId
     * @param queue
     */
    public ChannelDistributer(ZooKeeper zk, String hostId) {
        Preconditions.checkArgument(
                hostId != null && !hostId.trim().isEmpty(),
                "hostId is null or empty"
                );
        m_hostId = hostId;
        m_zk = Preconditions.checkNotNull(zk, "zookeeper is null");
        m_es = CoreUtils.getCachedSingleThreadExecutor("Import Channel Distributer for Host " + hostId, 15000);
        m_buses = CoreUtils.getCachedSingleThreadExecutor(
                "Import Channel Distributer Event Bus Dispatcher for Host " + hostId, 15000
                );
        m_eb = new AsyncEventBus(m_buses, eventBusFaultHandler);
        m_eb.register(this);
        m_mode = new AtomicStampedReference<>(OperationMode.RUNNING, 0);
        m_undispatched = new LinkedList<>();

        // Prime directory structure if needed
        OperationMode startMode = VoltDB.instance().getStartMode();
        assert startMode == OperationMode.RUNNING || startMode == OperationMode.PAUSED;
        mkdirs(zk, VoltZK.operationMode, startMode.getBytes());
        mkdirs(zk, HOST_DN, EMPTY_ARRAY);
        mkdirs(zk, MASTER_DN, EMPTY_ARRAY);

        GetOperationMode opMode = new GetOperationMode(VoltZK.operationMode);
        CreateNode createHostNode = new CreateNode(
                joinZKPath(HOST_DN, hostId),
                EMPTY_ARRAY, CreateMode.EPHEMERAL
                );
        MonitorHostNodes monitor = new MonitorHostNodes(HOST_DN);
        CreateNode electionCandidate = new CreateNode(
                CANDIDATE_PN,
                EMPTY_ARRAY, CreateMode.EPHEMERAL_SEQUENTIAL
                );
        ElectLeader elector = new ElectLeader(MASTER_DN, electionCandidate);

        createHostNode.getNode();
        elector.elect();

        m_candidate = electionCandidate.getNode();
        opMode.getMode();
        monitor.getChildren();
        // monitor the master list
        new GetChannels(MASTER_DN).getChannels();
    }

    public String getHostId() {
        return m_hostId;
    }

    public VersionedOperationMode getOperationMode() {
        return new VersionedOperationMode(m_mode);
    }

    /**
     * Register channels for the given importer. If they match to what is already registered
     * then nothing is done. Before registering channels, you need to register a callback
     * handler for channel assignments {@link #registerCallback(String, ChannelChangeCallback)}
     *
     * @param importer importer designation
     * @param uris list of channel URIs
     */
    public void registerChannels(String importer, Set<URI> uris) {
        NavigableSet<String> registered = m_callbacks.getReference().navigableKeySet();
        Preconditions.checkArgument(
                importer != null && !importer.trim().isEmpty(),
                "importer is null or empty"
                );
        Preconditions.checkArgument(uris != null, "uris set is null");
        Preconditions.checkArgument(
                !FluentIterable.from(uris).anyMatch(isNull()),
                "uris set %s contains null elements", uris
                );
        if (!registered.contains(importer)) {
            if (uris.isEmpty()) {
                // ImporterLifeCycleManager.stop() calls registerChannels() is called with an empty set of URIs.
                // If the importer never finished starting, we hit this condition.
                // This log message is used by the TestImporterStopAfterIncompleteStart JUnit.
                LOG.info("Skipping channel un-registration for " + importer + " since it did not finish initialization");
                return;
            } else {
                throw new IllegalStateException("no callbacks registered for " + importer
                        + " - unable to register channels " + Arrays.toString(uris.toArray()));
            }
        }

        Predicate<ChannelSpec> forImporter = ChannelSpec.importerIs(importer);
        Function<URI,ChannelSpec> asSpec = ChannelSpec.fromUri(importer);

        // convert method parameters to a set of ChannelSpecs
        NavigableSet<ChannelSpec> proposed = ImmutableSortedSet.copyOf(
                FluentIterable.from(uris).transform(asSpec)
                );

        LOG.info("(" + m_hostId + ") proposing channels " + proposed);

        int [] stamp = new int[]{0};

        ImmutableSortedSet.Builder<ChannelSpec> sbldr = null;
        NavigableSet<ChannelSpec> prev = null;
        SetData setter = null;
        NavigableSet<ChannelSpec> masterList = null;
        // retry writes when merging with stale data
        do {
            prev = m_channels.get(stamp);

            NavigableSet<ChannelSpec> current  = Sets.filter(prev, forImporter);
            if (current.equals(proposed)) {
                return;
            }
            sbldr = ImmutableSortedSet.naturalOrder();
            sbldr.addAll(Sets.filter(prev, not(forImporter)));
            sbldr.addAll(proposed);

            byte [] data = null;
            try {
                masterList = sbldr.build();
                data = asHostData(masterList);
            } catch (JSONException|IllegalArgumentException e) {
                throw loggedDistributerException(e, "failed to serialize the registration as json");
            }

            setter = new SetData(MASTER_DN, stamp[0], data);
        } while (setter.getCallbackCode() == Code.BADVERSION);

        //synch master channel list after channel registrations.
        int [] sstamp = new int[]{0};
        prev = m_channels.get(sstamp);
        m_channels.compareAndSet(prev, masterList, sstamp[0], stamp[0]);

        setter.getStat();
    }

    /**
     * Registers a (@link ChannelChangeCallback} for the given importer.
     * @param importer
     * @param callback a (@link ChannelChangeCallback}
     */
    public void registerCallback(String importer, ChannelChangeCallback callback) {
        Preconditions.checkArgument(
                importer != null && !importer.trim().isEmpty(),
                "importer is null or empty"
                );
        callback = checkNotNull(callback, "callback is null");

        if (m_done.get()) return;

        int [] stamp = new int[]{0};
        NavigableMap<String,ChannelChangeCallback> prev = null;
        NavigableMap<String,ChannelChangeCallback> next = null;
        ImmutableSortedMap.Builder<String,ChannelChangeCallback> mbldr = null;

        synchronized (m_undispatched) {
            do {
                prev = m_callbacks.get(stamp);
                mbldr = ImmutableSortedMap.naturalOrder();
                mbldr.putAll(Maps.filterKeys(prev, not(equalTo(importer))));
{'fraction_non_alphanumeric': 0.08976636877138083, 'fraction_numerical': 0.0019358773767535601, 'mean_word_length': 3.0151192504258946, 'pattern_counts': {'":': 2, '<': 77, '<?xml version=': 0, '>': 75, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '824462', 'n_tokens_mistral': 10536, 'n_tokens_neox': 10242, 'n_words': 2754}