<commit_before>package org.komamitsu.fluency.flusher;

import org.komamitsu.fluency.buffer.Buffer;
import org.komamitsu.fluency.sender.Sender;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.Flushable;
import java.io.IOException;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;

public abstract class Flusher
        implements Flushable, Closeable
{
    private static final Logger LOG = LoggerFactory.getLogger(Flusher.class);
    private final AtomicBoolean isTerminated = new AtomicBoolean();
    protected final Buffer buffer;
    protected final Sender sender;
    private final Config config;

    protected Flusher(Buffer buffer, Sender sender, Config config)
    {
        this.buffer = buffer;
        this.sender = sender;
        this.config = config;
    }

    public Buffer getBuffer()
    {
        return buffer;
    }

    protected abstract void flushInternal(boolean force)
            throws IOException;

    protected abstract void beforeClosingBuffer()
            throws IOException;

    public void onUpdate()
            throws IOException
    {
        flushInternal(false);
    }

    @Override
    public void flush()
            throws IOException
    {
        flushInternal(true);
    }

    @Override
    public void close()
            throws IOException
    {
        try {
            beforeClosingBuffer();
        }
        catch (Exception e) {
            LOG.error("Failed to call beforeClosingBuffer()", e);
        }
        finally {
            try {
                sender.close();
            }
            catch (Exception e) {
                LOG.error("Failed to close the sender", e);
            }
            finally {
                ExecutorService executor = Executors.newSingleThreadExecutor();
                Future<Void> future = executor.submit(new Callable<Void>()
                {
                    @Override
                    public Void call()
                            throws Exception
                    {
                        closeBuffer();
                        isTerminated.set(true);
                        return null;
                    }
                });

                try {
                    future.get(config.getWaitUntilTerminated(), TimeUnit.SECONDS);
                }
                catch (InterruptedException e) {
                    LOG.warn("Interrupted", e);
                    Thread.currentThread().interrupt();
                }
                catch (ExecutionException e) {
                    LOG.warn("closeBuffer() failed", e);
                }
                catch (TimeoutException e) {
                    LOG.warn("closeBuffer() timed out", e);
                }
            }
        }
    }

    public boolean isTerminated()
    {
        return isTerminated.get();
    }

    private void closeBuffer()
    {
        LOG.trace("closeBuffer(): closing buffer");
        buffer.close();
    }

    public Sender getSender()
    {
        return sender;
    }

    public int getFlushIntervalMillis()
    {
        return config.getFlushIntervalMillis();
    }

    public int getWaitUntilBufferFlushed()
    {
        return config.getWaitUntilBufferFlushed();
    }

    public int getWaitUntilTerminated()
    {
        return config.getWaitUntilTerminated();
    }

    @Override
    public String toString()
    {
        return "Flusher{" +
                "isTerminated=" + isTerminated +
                ", buffer=" + buffer +
                ", sender=" + sender +
                ", config=" + config +
                '}';
    }

    public static class Config
    {
        private int flushIntervalMillis = 600;
        private int waitUntilBufferFlushed = 10;
        private int waitUntilTerminated = 10;

        public int getFlushIntervalMillis()
        {
            return flushIntervalMillis;
        }

        public Config setFlushIntervalMillis(int flushIntervalMillis)
        {
            this.flushIntervalMillis = flushIntervalMillis;
            return this;
        }

        public int getWaitUntilBufferFlushed()
        {
            return waitUntilBufferFlushed;
        }

        public Config setWaitUntilBufferFlushed(int waitUntilBufferFlushed)
        {
            this.waitUntilBufferFlushed = waitUntilBufferFlushed;
            return this;
        }

        public int getWaitUntilTerminated()
        {
            return waitUntilTerminated;
        }

        public Config setWaitUntilTerminated(int waitUntilTerminated)
        {
            this.waitUntilTerminated = waitUntilTerminated;
            return this;
        }

        @Override
        public String toString()
        {
            return "Config{" +
                    "flushIntervalMillis=" + flushIntervalMillis +
                    ", waitUntilBufferFlushed=" + waitUntilBufferFlushed +
                    ", waitUntilTerminated=" + waitUntilTerminated +
                    '}';
        }
    }

    public interface Instantiator
    {
        Flusher createInstance(Buffer buffer, Sender sender);
    }
}
<commit_msg>Bug fixed: executor service not terminated<commit_after>package org.komamitsu.fluency.flusher;

import org.komamitsu.fluency.buffer.Buffer;
import org.komamitsu.fluency.sender.Sender;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.Flushable;
import java.io.IOException;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;

public abstract class Flusher
        implements Flushable, Closeable
{
    private static final Logger LOG = LoggerFactory.getLogger(Flusher.class);
    private final AtomicBoolean isTerminated = new AtomicBoolean();
    protected final Buffer buffer;
    protected final Sender sender;
    private final Config config;

    protected Flusher(Buffer buffer, Sender sender, Config config)
    {
        this.buffer = buffer;
        this.sender = sender;
        this.config = config;
    }

    public Buffer getBuffer()
    {
        return buffer;
    }

    protected abstract void flushInternal(boolean force)
            throws IOException;

    protected abstract void beforeClosingBuffer()
            throws IOException;

    public void onUpdate()
            throws IOException
    {
        flushInternal(false);
    }

    @Override
    public void flush()
            throws IOException
    {
        flushInternal(true);
    }

    @Override
    public void close()
            throws IOException
    {
        try {
            beforeClosingBuffer();
        }
        catch (Exception e) {
            LOG.error("Failed to call beforeClosingBuffer()", e);
        }
        finally {
            try {
                sender.close();
            }
            catch (Exception e) {
                LOG.error("Failed to close the sender", e);
            }
            finally {
                ExecutorService executor = Executors.newSingleThreadExecutor();
                Future<Void> future = executor.submit(new Callable<Void>()
                {
                    @Override
                    public Void call()
                            throws Exception
                    {
                        closeBuffer();
                        isTerminated.set(true);
                        return null;
                    }
                });

                try {
                    future.get(config.getWaitUntilTerminated(), TimeUnit.SECONDS);
                }
                catch (InterruptedException e) {
                    LOG.warn("Interrupted", e);
                    Thread.currentThread().interrupt();
                }
                catch (ExecutionException e) {
                    LOG.warn("closeBuffer() failed", e);
                }
                catch (TimeoutException e) {
                    LOG.warn("closeBuffer() timed out", e);
                } finally {
                    executor.shutdown();
                }
            }
        }
    }

    public boolean isTerminated()
    {
        return isTerminated.get();
    }

    private void closeBuffer()
    {
        LOG.trace("closeBuffer(): closing buffer");
        buffer.close();
    }

    public Sender getSender()
    {
        return sender;
    }

    public int getFlushIntervalMillis()
    {
        return config.getFlushIntervalMillis();
    }

    public int getWaitUntilBufferFlushed()
    {
        return config.getWaitUntilBufferFlushed();
    }

    public int getWaitUntilTerminated()
    {
        return config.getWaitUntilTerminated();
    }

    @Override
    public String toString()
    {
        return "Flusher{" +
                "isTerminated=" + isTerminated +
                ", buffer=" + buffer +
                ", sender=" + sender +
                ", config=" + config +
                '}';
    }

    public static class Config
    {
        private int flushIntervalMillis = 600;
        private int waitUntilBufferFlushed = 10;
        private int waitUntilTerminated = 10;

        public int getFlushIntervalMillis()
        {
            return flushIntervalMillis;
        }

        public Config setFlushIntervalMillis(int flushIntervalMillis)
        {
            this.flushIntervalMillis = flushIntervalMillis;
            return this;
        }

        public int getWaitUntilBufferFlushed()
        {
            return waitUntilBufferFlushed;
        }

        public Config setWaitUntilBufferFlushed(int waitUntilBufferFlushed)
        {
            this.waitUntilBufferFlushed = waitUntilBufferFlushed;
            return this;
        }

        public int getWaitUntilTerminated()
        {
            return waitUntilTerminated;
        }

        public Config setWaitUntilTerminated(int waitUntilTerminated)
        {
            this.waitUntilTerminated = waitUntilTerminated;
            return this;
        }

        @Override
        public String toString()
        {
            return "Config{" +
                    "flushIntervalMillis=" + flushIntervalMillis +
                    ", waitUntilBufferFlushed=" + waitUntilBufferFlushed +
                    ", waitUntilTerminated=" + waitUntilTerminated +
                    '}';
        }
    }

    public interface Instantiator
    {
        Flusher createInstance(Buffer buffer, Sender sender);
    }
}

{'fraction_non_alphanumeric': 0.07836503291880029, 'fraction_numerical': 0.0016459400146305778, 'mean_word_length': 1.669514278740542, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, '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': '2010954', 'n_tokens_mistral': 2737, 'n_tokens_neox': 2630, 'n_words': 599}