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

import com.google.common.base.Charsets;
import com.google.common.collect.Lists;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreMutex;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.data.Stat;
import org.echocat.jomon.runtime.concurrent.RetryForSpecifiedCountStrategy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zalando.nakadi.domain.EventTypePartition;
import org.zalando.nakadi.exceptions.NakadiRuntimeException;
import org.zalando.nakadi.exceptions.ServiceUnavailableException;
import org.zalando.nakadi.exceptions.UnableProcessException;
import org.zalando.nakadi.exceptions.runtime.MyNakadiRuntimeException1;
import org.zalando.nakadi.exceptions.runtime.OperationInterruptedException;
import org.zalando.nakadi.exceptions.runtime.OperationTimeoutException;
import org.zalando.nakadi.exceptions.runtime.RequestInProgressException;
import org.zalando.nakadi.exceptions.runtime.ZookeeperException;
import org.zalando.nakadi.service.subscription.model.Session;
import org.zalando.nakadi.view.SubscriptionCursorWithoutToken;

import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

import static com.google.common.base.Charsets.UTF_8;
import static org.echocat.jomon.runtime.concurrent.Retryer.executeWithRetry;

public abstract class AbstractZkSubscriptionClient implements ZkSubscriptionClient {
    private static final String STATE_INITIALIZED = "INITIALIZED";
    private static final int COMMIT_CONFLICT_RETRY_TIMES = 5;
    protected static final String NODE_TOPOLOGY = "/topology";

    private final String subscriptionId;
    private final CuratorFramework curatorFramework;
    private InterProcessSemaphoreMutex lock;
    private final String resetCursorPath;
    private final Logger log;

    public AbstractZkSubscriptionClient(
            final String subscriptionId,
            final CuratorFramework curatorFramework,
            final String loggingPath) {
        this.subscriptionId = subscriptionId;
        this.curatorFramework = curatorFramework;
        this.resetCursorPath = getSubscriptionPath("/cursor_reset");
        this.log = LoggerFactory.getLogger(loggingPath + ".zk");
    }

    protected CuratorFramework getCurator() {
        return this.curatorFramework;
    }

    protected String getSubscriptionId() {
        return subscriptionId;
    }

    protected String getSubscriptionPath(final String value) {
        return "/nakadi/subscriptions/" + subscriptionId + value;
    }

    protected Logger getLog() {
        return log;
    }

    @Override
    public final void runLocked(final Runnable function) {
        try {
            Exception releaseException = null;
            if (null == lock) {
                lock = new InterProcessSemaphoreMutex(curatorFramework, "/nakadi/locks/subscription_" + subscriptionId);
            }

            lock.acquire();
            try {
                function.run();
            } finally {
                try {
                    lock.release();
                } catch (final Exception e) {
                    log.error("Failed to release lock", e);
                    releaseException = e;
                }
            }
            if (releaseException != null) {
                throw releaseException;
            }
        } catch (final NakadiRuntimeException | MyNakadiRuntimeException1 e) {
            throw e;
        } catch (final Exception e) {
            throw new NakadiRuntimeException(e);
        }
    }<commit_msg>ARUHA-1388 Reduce the impact of leaving lock for subscription, and be able to filter out this kind of subscriptions.<commit_after>package org.zalando.nakadi.service.subscription.zk;

import com.google.common.base.Charsets;
import com.google.common.collect.Lists;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreMutex;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.data.Stat;
import org.echocat.jomon.runtime.concurrent.RetryForSpecifiedCountStrategy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zalando.nakadi.domain.EventTypePartition;
import org.zalando.nakadi.exceptions.NakadiRuntimeException;
import org.zalando.nakadi.exceptions.ServiceUnavailableException;
import org.zalando.nakadi.exceptions.UnableProcessException;
import org.zalando.nakadi.exceptions.runtime.MyNakadiRuntimeException1;
import org.zalando.nakadi.exceptions.runtime.OperationInterruptedException;
import org.zalando.nakadi.exceptions.runtime.OperationTimeoutException;
import org.zalando.nakadi.exceptions.runtime.RequestInProgressException;
import org.zalando.nakadi.exceptions.runtime.ZookeeperException;
import org.zalando.nakadi.service.subscription.model.Session;
import org.zalando.nakadi.view.SubscriptionCursorWithoutToken;

import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

import static com.google.common.base.Charsets.UTF_8;
import static org.echocat.jomon.runtime.concurrent.Retryer.executeWithRetry;

public abstract class AbstractZkSubscriptionClient implements ZkSubscriptionClient {
    private static final String STATE_INITIALIZED = "INITIALIZED";
    private static final int COMMIT_CONFLICT_RETRY_TIMES = 5;
    protected static final String NODE_TOPOLOGY = "/topology";
    public static final int SECONDS_TO_WAIT_FOR_LOCK = 15;

    private final String subscriptionId;
    private final CuratorFramework curatorFramework;
    private InterProcessSemaphoreMutex lock;
    private final String resetCursorPath;
    private final Logger log;

    public AbstractZkSubscriptionClient(
            final String subscriptionId,
            final CuratorFramework curatorFramework,
            final String loggingPath) {
        this.subscriptionId = subscriptionId;
        this.curatorFramework = curatorFramework;
        this.resetCursorPath = getSubscriptionPath("/cursor_reset");
        this.log = LoggerFactory.getLogger(loggingPath + ".zk");
    }

    protected CuratorFramework getCurator() {
        return this.curatorFramework;
    }

    protected String getSubscriptionId() {
        return subscriptionId;
    }

    protected String getSubscriptionPath(final String value) {
        return "/nakadi/subscriptions/" + subscriptionId + value;
    }

    protected Logger getLog() {
        return log;
    }

    @Override
    public final void runLocked(final Runnable function) {
        try {
            Exception releaseException = null;
            if (null == lock) {
                lock = new InterProcessSemaphoreMutex(curatorFramework, "/nakadi/locks/subscription_" + subscriptionId);
            }

            final boolean acquired = lock.acquire(SECONDS_TO_WAIT_FOR_LOCK, TimeUnit.SECONDS);
            if (!acquired) {
                throw new ServiceUnavailableException("Failed to acquire subscription lock within " +
                        SECONDS_TO_WAIT_FOR_LOCK + " seconds");
            }
            try {
                function.run();
            } finally {
                try {
                    lock.release();
                } catch (final Exception e) {
                    log.error("Failed to release lock", e);
                    releaseException = e;
                }
            }
            if (releaseException != null) {
                throw releaseException;
            }
        } catch (final NakadiRuntimeException | MyNakadiRuntimeException1 e) {
            throw e;
        } catch (final Exception e) {
            throw new NakadiRuntimeException(e);
        }
    }
{'fraction_non_alphanumeric': 0.08028857342331859, 'fraction_numerical': 0.002094484524086572, 'mean_word_length': 3.282511210762332, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1121059', 'n_tokens_mistral': 2345, 'n_tokens_neox': 2326, 'n_words': 482}