<commit_before>package org.zalando.nakadi.service.subscription.zk;

import com.fasterxml.jackson.annotation.JsonProperty;
import org.apache.commons.codec.binary.Hex;
import org.zalando.nakadi.domain.EventTypePartition;
import org.zalando.nakadi.exceptions.runtime.NakadiBaseException;
import org.zalando.nakadi.exceptions.runtime.NakadiRuntimeException;
import org.zalando.nakadi.exceptions.runtime.OperationTimeoutException;
import org.zalando.nakadi.exceptions.runtime.ServiceTemporarilyUnavailableException;
import org.zalando.nakadi.exceptions.runtime.ZookeeperException;
import org.zalando.nakadi.service.subscription.model.Partition;
import org.zalando.nakadi.service.subscription.model.Session;
import org.zalando.nakadi.view.SubscriptionCursorWithoutToken;

import javax.annotation.Nullable;
import java.io.Closeable;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.Arrays;
import java.util.Collection;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.Callable;

public interface ZkSubscriptionClient extends Closeable {

    /**
     * Makes runLocked on subscription, using zk path /nakadi/locks/subscription_{subscriptionId}
     * Lock is created as an ephemeral node, so it will be deleted if nakadi goes down. After obtaining runLocked,
     * provided function will be called under subscription runLocked
     *
     * @param function Function to call in context of runLocked.
     */
    <T> T runLocked(Callable<T> function);

    default void runLocked(final Runnable function) {
        runLocked((Callable<Void>) () -> {
            function.run();
            return null;
        });
    }


    /**
     * Creates subscription node in zookeeper on path /nakadi/subscriptions/{subscriptionId}
     *
     * @return true if subscription was created. False if subscription already present. To operate on this value
     * additional field 'state' is used /nakadi/subscriptions/{subscriptionId}/state. Just after creation it has value
     * CREATED. After {{@link #fillEmptySubscription}} call it will have value INITIALIZED. So true
     * will be returned in case of state is equal to CREATED.
     */
    boolean isSubscriptionCreatedAndInitialized() throws NakadiRuntimeException;

    /**
     * Deletes subscription with all its data in zookeeper
     */
    void deleteSubscription();

    /**
     * Fills subscription object using partitions information provided (mapping from (topic, partition) to real offset
     * (NOT "BEGIN").
     *
     * @param cursors Data to use for subscription filling.
     */
    void fillEmptySubscription(Collection<SubscriptionCursorWithoutToken> cursors);

    /**
     * Updates specified partitions in zk.
     */
    void updatePartitionsConfiguration(String newSessionsHash, Partition[] partitions) throws NakadiRuntimeException,
            SubscriptionNotInitializedException;

    /**
     * Returns session list in zk related to this subscription.
     *
     * @return List of existing sessions.
     */
    Collection<Session> listSessions()
            throws SubscriptionNotInitializedException, NakadiRuntimeException, ServiceTemporarilyUnavailableException;

    boolean isActiveSession(String streamId) throws ServiceTemporarilyUnavailableException;

    /**
     * List partitions
     *
     * @return list of partitions related to this subscription.
     */
    Topology getTopology() throws SubscriptionNotInitializedException, NakadiRuntimeException;

    /**
     * Subscribes to changes of session list in /nakadi/subscriptions/{subscriptionId}/sessions.
     * Each time list is changed calls {{@code listener}} subscription.
     *
     * @param listener method to call on any change of client list.
     */
    ZkSubscription<List<String>> subscribeForSessionListChanges(Runnable listener) throws NakadiRuntimeException;

    /**
     * Subscribe for topology changes.
     *
     * @param listener called whenever /nakadi/subscriptions/{subscriptionId}/topology node is changed.
     * @return Subscription instance
     */
    ZkSubscription<Topology> subscribeForTopologyChanges(Runnable listener) throws NakadiRuntimeException;

    ZkSubscription<SubscriptionCursorWithoutToken> subscribeForOffsetChanges(
            EventTypePartition key, Runnable commitListener);

    /**
     * Returns committed offset values for specified partition keys.
     * Offsets include timeline and version data.
     * The value that is stored there is a view value, so it will look like
     * 001-0001-00000000000000000001
     *
     * @param keys Key to get offset for
     * @return commit offset
     */
    Map<EventTypePartition, SubscriptionCursorWithoutToken> getOffsets(Collection<EventTypePartition> keys)
            throws NakadiRuntimeException;

    /**
     * Forcefully set commits offsets specified in {@link SubscriptionCursorWithoutToken}
     *
     * @param cursors - offsets to set for subscription
     * @throws Exception
     */
    void forceCommitOffsets(final List<SubscriptionCursorWithoutToken> cursors) throws NakadiRuntimeException;

    List<Boolean> commitOffsets(List<SubscriptionCursorWithoutToken> cursors,
                                Comparator<SubscriptionCursorWithoutToken> comparator);

    /**
     * Registers client connection using session id in /nakadi/subscriptions/{subscriptionId}/sessions/{session.id}
     * and value {{session.weight}}
     *
     * @param session Session to register.
     */<commit_msg>close subscription connections<commit_after>package org.zalando.nakadi.service.subscription.zk;

import com.fasterxml.jackson.annotation.JsonProperty;
import org.apache.commons.codec.binary.Hex;
import org.zalando.nakadi.domain.EventTypePartition;
import org.zalando.nakadi.exceptions.runtime.NakadiBaseException;
import org.zalando.nakadi.exceptions.runtime.NakadiRuntimeException;
import org.zalando.nakadi.exceptions.runtime.OperationTimeoutException;
import org.zalando.nakadi.exceptions.runtime.ServiceTemporarilyUnavailableException;
import org.zalando.nakadi.exceptions.runtime.ZookeeperException;
import org.zalando.nakadi.service.subscription.model.Partition;
import org.zalando.nakadi.service.subscription.model.Session;
import org.zalando.nakadi.view.SubscriptionCursorWithoutToken;

import javax.annotation.Nullable;
import java.io.Closeable;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.Arrays;
import java.util.Collection;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.Callable;

public interface ZkSubscriptionClient extends Closeable {

    /**
     * Makes runLocked on subscription, using zk path /nakadi/locks/subscription_{subscriptionId}
     * Lock is created as an ephemeral node, so it will be deleted if nakadi goes down. After obtaining runLocked,
     * provided function will be called under subscription runLocked
     *
     * @param function Function to call in context of runLocked.
     */
    <T> T runLocked(Callable<T> function);

    default void runLocked(final Runnable function) {
        runLocked((Callable<Void>) () -> {
            function.run();
            return null;
        });
    }


    /**
     * Creates subscription node in zookeeper on path /nakadi/subscriptions/{subscriptionId}
     *
     * @return true if subscription was created. False if subscription already present. To operate on this value
     * additional field 'state' is used /nakadi/subscriptions/{subscriptionId}/state. Just after creation it has value
     * CREATED. After {{@link #fillEmptySubscription}} call it will have value INITIALIZED. So true
     * will be returned in case of state is equal to CREATED.
     */
    boolean isSubscriptionCreatedAndInitialized() throws NakadiRuntimeException;

    /**
     * Deletes subscription with all its data in zookeeper
     */
    void deleteSubscription();

    /**
     * Fills subscription object using partitions information provided (mapping from (topic, partition) to real offset
     * (NOT "BEGIN").
     *
     * @param cursors Data to use for subscription filling.
     */
    void fillEmptySubscription(Collection<SubscriptionCursorWithoutToken> cursors);

    /**
     * Updates specified partitions in zk.
     */
    void updatePartitionsConfiguration(String newSessionsHash, Partition[] partitions) throws NakadiRuntimeException,
            SubscriptionNotInitializedException;

    /**
     * Returns session list in zk related to this subscription.
     *
     * @return List of existing sessions.
     */
    Collection<Session> listSessions()
            throws SubscriptionNotInitializedException, NakadiRuntimeException, ServiceTemporarilyUnavailableException;

    boolean isActiveSession(String streamId) throws ServiceTemporarilyUnavailableException;

    /**
     * List partitions
     *
     * @return list of partitions related to this subscription.
     */
    Topology getTopology() throws SubscriptionNotInitializedException, NakadiRuntimeException;

    /**
     * Subscribes to changes of session list in /nakadi/subscriptions/{subscriptionId}/sessions.
     * Each time list is changed calls {{@code listener}} subscription.
     *
     * @param listener method to call on any change of client list.
     */
    ZkSubscription<List<String>> subscribeForSessionListChanges(Runnable listener) throws NakadiRuntimeException;

    /**
     * Subscribe for topology changes.
     *
     * @param listener called whenever /nakadi/subscriptions/{subscriptionId}/topology node is changed.
     * @return Subscription instance
     */
    ZkSubscription<Topology> subscribeForTopologyChanges(Runnable listener) throws NakadiRuntimeException;

    ZkSubscription<SubscriptionCursorWithoutToken> subscribeForOffsetChanges(
            EventTypePartition key, Runnable commitListener);

    /**
     * Returns committed offset values for specified partition keys.
     * Offsets include timeline and version data.
     * The value that is stored there is a view value, so it will look like
     * 001-0001-00000000000000000001
     *
     * @param keys Key to get offset for
     * @return commit offset
     */
    Map<EventTypePartition, SubscriptionCursorWithoutToken> getOffsets(Collection<EventTypePartition> keys)
            throws NakadiRuntimeException;

    /**
     * Forcefully set commits offsets specified in {@link SubscriptionCursorWithoutToken}
     *
     * @param cursors - offsets to set for subscription
     * @throws Exception
     */
    void forceCommitOffsets(List<SubscriptionCursorWithoutToken> cursors) throws NakadiRuntimeException;

    List<Boolean> commitOffsets(List<SubscriptionCursorWithoutToken> cursors,
                                Comparator<SubscriptionCursorWithoutToken> comparator);

    /**
     * Registers client connection using session id in /nakadi/subscriptions/{subscriptionId}/sessions/{session.id}
     * and value {{session.weight}}
     *
     * @param session Session to register.
     */
{'fraction_non_alphanumeric': 0.07651832694194238, 'fraction_numerical': 0.004815838758583786, 'mean_word_length': 3.7038590604026846, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 2, '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': '422774', 'n_tokens_mistral': 2928, 'n_tokens_neox': 2857, 'n_words': 825}