public class Fluency
        implements Flushable, Closeable
{
    private static final Logger LOG = LoggerFactory.getLogger(Fluency.class);
    private final Buffer buffer;
    private final Flusher flusher;
    private final Emitter emitter = new Emitter();

    // Fluency has this public constructor, but using FluentBuilder is usually more recommended.
    public Fluency(Buffer buffer, Flusher flusher)
    {
        this.buffer = buffer;
        this.flusher = flusher;
    }

    public void emit(final String tag, final long timestamp, final Map<String, Object> data)
            throws IOException
    {
        emitter.emit(() -> buffer.append(tag, timestamp, data));
    }

    public void emit(String tag, Map<String, Object> data)
            throws IOException
    {
        emit(tag, System.currentTimeMillis() / 1000, data);
    }

    public void emit(final String tag, final EventTime eventTime, final Map<String, Object> data)
            throws IOException
    {
        emitter.emit(() -> buffer.append(tag, eventTime, data));
    }

    public void emit(final String tag, final long timestamp, final byte[] mapValue, final int offset, final int len)
            throws IOException
    {
        emitter.emit(() -> buffer.appendMessagePackMapValue(tag, timestamp, mapValue, offset, len));
    }

    public void emit(String tag, byte[] mapValue, int offset, int len)
            throws IOException
    {
        emit(tag, System.currentTimeMillis() / 1000, mapValue, offset, len);
    }

    public void emit(final String tag, final EventTime eventTime, final byte[] mapValue, final int offset, final int len)
            throws IOException
    {
        emitter.emit(() -> buffer.appendMessagePackMapValue(tag, eventTime, mapValue, offset, len));
    }

    public void emit(final String tag, final long timestamp, final ByteBuffer mapValue)
            throws IOException
    {
        emitter.emit(() -> buffer.appendMessagePackMapValue(tag, timestamp, mapValue));
    }

    public void emit(String tag, ByteBuffer mapValue)
            throws IOException
    {
        emit(tag, System.currentTimeMillis() / 1000, mapValue);
    }

    public void emit(final String tag, final EventTime eventTime, final ByteBuffer mapValue)
            throws IOException
    {
        emitter.emit(() -> buffer.appendMessagePackMapValue(tag, eventTime, mapValue));
    }

    @Override
    public void flush()
            throws IOException
    {
        flusher.flush();
    }

    @Override
    public void close()
            throws IOException
    {
        flusher.close();
    }

    public void clearBackupFiles()
    {
        buffer.clearBackupFiles();
    }

    public long getAllocatedBufferSize()
    {
        return buffer.getAllocatedSize();
    }

    public long getBufferedDataSize()
    {
        return buffer.getBufferedDataSize();
    }

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

    public boolean waitUntilAllBufferFlushed(int maxWaitSeconds)
            throws InterruptedException
    {
        int intervalMilli = 500;
        for (int i = 0; i < maxWaitSeconds * (1000 / intervalMilli); i++) {
            long bufferedDataSize = getBufferedDataSize();
            LOG.debug("Waiting for flushing all buffer: {}", bufferedDataSize);
            if (bufferedDataSize == 0) {
                return true;
            }
            TimeUnit.MILLISECONDS.sleep(intervalMilli);
        }
        LOG.warn("Buffered data still remains: {}", getBufferedDataSize());
        return false;
    }

    public boolean waitUntilFlusherTerminated(int maxWaitSeconds)
            throws InterruptedException
    {
        int intervalMilli = 500;
        for (int i = 0; i < maxWaitSeconds * (1000 / intervalMilli); i++) {
            boolean terminated = isTerminated();
            LOG.debug("Waiting until the flusher is terminated: {}", terminated);
            if (terminated) {
                return true;
            }
            TimeUnit.MILLISECONDS.sleep(intervalMilli);
        }
        LOG.warn("The flusher isn't terminated");
        return false;
    }

    public Buffer getBuffer()
    {
        return buffer;
    }

    public Flusher getFlusher()
    {
        return flusher;
    }

    @Override
    public String toString()
    {
        return "Fluency{" +
                "buffer=" + buffer +
                ", flusher=" + flusher +
                '}';
    }

    private interface Append
    {
        void append()
                throws IOException;
    }

    private class Emitter
    {
        void emit(Append appender)
                throws IOException
        {
            try {
                appender.append();
            }
            catch (BufferFullException e) {
                LOG.error("emit() failed due to buffer full. Flushing buffer. Please try again...");
                flusher.flush();
                throw e;
            }
        }
    }
} public class FluentdIngester
        implements Ingester
{
    private final Config config;
    private final FluentdSender sender;
    private final ObjectMapper objectMapper = new ObjectMapper(new MessagePackFactory());

    public FluentdIngester(FluentdSender sender)
    {
        this(new Config(), sender);
    }

    public FluentdIngester(Config config, FluentdSender sender)
    {
        this.config = config;
        this.sender = sender;
    }

    @Override
    public void ingest(String tag, ByteBuffer dataBuffer)
            throws IOException
    {
        ByteArrayOutputStream header = new ByteArrayOutputStream();
        MessagePacker messagePacker = MessagePack.newDefaultPacker(header);

        int dataLength = dataBuffer.limit();
        messagePacker.packArrayHeader(3);
        messagePacker.packString(tag);
        messagePacker.packRawStringHeader(dataLength);
        messagePacker.flush();

        ByteBuffer headerBuffer = ByteBuffer.wrap(header.toByteArray());

        if (config.isAckResponseMode()) {
            // The spec (https://github.com/fluent/fluentd/wiki/Forward-Protocol-Specification-v1#entry)
            // says to encode a 128 bit value as base64, but fluent-bit currently does something different
            // and in fluent-bit and fluentd there is no validation. So for now we keep it simple, see
            // discussion on issue #181.
            String token = UUID.randomUUID().toString();

            ByteBuffer optionBuffer = ByteBuffer.wrap(objectMapper.writeValueAsBytes(new RequestOption(dataLength, token)));
            List<ByteBuffer> buffers = Arrays.asList(headerBuffer, dataBuffer, optionBuffer);

            synchronized (sender) {
                sender.sendWithAck(buffers, token);
            }
        }
        else {
            ByteBuffer optionBuffer = ByteBuffer.wrap(objectMapper.writeValueAsBytes(new RequestOption(dataLength, null)));
            List<ByteBuffer> buffers = Arrays.asList(headerBuffer, dataBuffer, optionBuffer);

            synchronized (sender) {
                sender.send(buffers);
            }
        }
    }

    @Override
    public Sender getSender()
    {
        return sender;
    }

    public boolean isAckResponseMode()
    {
        return config.isAckResponseMode();
    }

    @Override
    public void close()
            throws IOException
    {
        sender.close();
    }

    public static class Config
    {
        private boolean ackResponseMode = false;

        public boolean isAckResponseMode()
        {
            return ackResponseMode;
        }

        public void setAckResponseMode(boolean ackResponseMode)
        {
            this.ackResponseMode = ackResponseMode;
        }
    }
} public abstract class FluentdSender
        implements Closeable, Sender
{
    private static final Logger LOG = LoggerFactory.getLogger(FluentdSender.class);
    private final Config config;

    protected FluentdSender(Config config)
    {
        this.config = config;
    }

    protected FluentdSender()
    {
        this(new FluentdSender.Config());
    }

    public synchronized void send(ByteBuffer buffer)
            throws IOException
    {
        sendInternalWithRestoreBufferPositions(Arrays.asList(buffer), null);
    }

    public synchronized void send(List<ByteBuffer> buffers)
            throws IOException
    {
        sendInternalWithRestoreBufferPositions(buffers, null);
    }

    public void sendWithAck(List<ByteBuffer> buffers, String ackToken)
            throws IOException
    {
        sendInternalWithRestoreBufferPositions(buffers, ackToken);
    }

    private void sendInternalWithRestoreBufferPositions(List<ByteBuffer> buffers, String ackToken)
            throws IOException
    {
        List<Integer> positions = new ArrayList<>(buffers.size());
        for (ByteBuffer data : buffers) {
            positions.add(data.position());
        }

        try {
            sendInternal(buffers, ackToken);
        }
        catch (Exception e) {
            for (int i = 0; i < buffers.size(); i++) {
                buffers.get(i).position(positions.get(i));
            }

            ErrorHandler errorHandler = config.getErrorHandler();
            if (errorHandler != null) {
                try {
                    errorHandler.handle(e);
                }
                catch (Exception ex) {
                    LOG.warn("Failed to handle an error in the error handler {}", errorHandler, ex);
                }
            }

            if (e instanceof IOException) {
                throw (IOException) e;
            }
            else {
                throw new IOException(e);
            }
        }
    }

    public abstract boolean isAvailable();

    abstract protected void sendInternal(List<ByteBuffer> buffers, String ackToken)
            throws IOException;

    public static class Config
            extends Sender.Config
    {
    }
} public class MultiSender
        extends FluentdSender
{
    private static final Logger LOG = LoggerFactory.getLogger(MultiSender.class);
    private final List<FluentdSender> senders;

    public MultiSender(List<FluentdSender> senders)
    {
        this(new Config(), senders);
    }

    public MultiSender(Config config, List<FluentdSender> senders)
    {
        super(config);
        this.senders = senders;
    }

    @Override
    public boolean isAvailable()
    {
        return true;
    }

    @Override
    protected synchronized void sendInternal(List<ByteBuffer> buffers, String ackToken)
            throws AllNodesUnavailableException
    {
        for (FluentdSender sender : senders) {
            boolean isAvailable = sender.isAvailable();
            LOG.trace("send(): sender={}, isAvailable={}", sender, isAvailable);
            if (isAvailable) {
                try {
                    if (ackToken == null) {
                        sender.send(buffers);
                    }
                    else {
                        sender.sendWithAck(buffers, ackToken);
                    }
                    return;
                }
                catch (IOException e) {
                    LOG.warn("Failed to send: sender=" + sender + ". Trying to use next sender...", e);
                }
            }
        }
        throw new AllNodesUnavailableException("All nodes are unavailable");
    }

    @Override
    public void close()
            throws IOException
    {
        IOException firstException = null;
        for (FluentdSender sender : senders) {
            try {
                sender.close();
            }
            catch (IOException e) {
                if (firstException == null) {
                    firstException = e;
                }
            }
        }
        if (firstException != null) {
            throw firstException;
        }
    }

    public List<FluentdSender> getSenders()
    {
        return Collections.unmodifiableList(senders);
    }

    @Override
    public String toString()
    {
        return "MultiSender{" +
                "senders=" + senders +
                "} " + super.toString();
    }

    public static class AllNodesUnavailableException
            extends IOException
    {
        public AllNodesUnavailableException(String s)
        {
            super(s);
        }
    }

    public static class Config
            extends FluentdSender.Config
    {
    }
} public class RetryableSender
        extends FluentdSender
{
    private static final Logger LOG = LoggerFactory.getLogger(RetryableSender.class);
    private final FluentdSender baseSender;
    private final AtomicBoolean isClosed = new AtomicBoolean();
    private RetryStrategy retryStrategy;

    public RetryableSender(FluentdSender baseSender, RetryStrategy retryStrategy)
    {
        this(new Config(), baseSender, retryStrategy);
    }

    public RetryableSender(Config config, FluentdSender baseSender, RetryStrategy retryStrategy)
    {
        super(config);
        this.baseSender = baseSender;
        this.retryStrategy = retryStrategy;
    }

    @Override
    public void close()
            throws IOException
    {
        baseSender.close();
        isClosed.set(true);
    }

    @Override
    public boolean isAvailable()
    {
        return true;
    }

    @Override
    protected synchronized void sendInternal(List<ByteBuffer> buffers, String ackToken)
            throws IOException
    {
        IOException firstException = null;

        int retry = 0;
        while (!retryStrategy.isRetriedOver(retry)) {
            if (isClosed.get()) {
                throw new RetryOverException("This sender is already closed", firstException);
            }

            try {
                if (ackToken == null) {
                    baseSender.send(buffers);
                }
                else {
                    baseSender.sendWithAck(buffers, ackToken);
                }
                return;
            }
            catch (IOException e) {
                firstException = e;
                LOG.warn("Sender failed to send data. sender=" + this + ", retry=" + retry, e);
            }

            try {
                TimeUnit.MILLISECONDS.sleep(retryStrategy.getNextIntervalMillis(retry));
            }
            catch (InterruptedException e) {
                LOG.debug("Interrupted while waiting", e);
                Thread.currentThread().interrupt();
            }
            retry++;
        }

        throw new RetryOverException("Sending data was retried over", firstException);
    }

    public FluentdSender getBaseSender()
    {
        return baseSender;
    }

    public RetryStrategy getRetryStrategy()
    {
        return retryStrategy;
    }

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

    @Override
    public String toString()
    {
        return "RetryableSender{" +
                "baseSender=" + baseSender +
                ", retryStrategy=" + retryStrategy +
                ", isClosed=" + isClosed +
                "} " + super.toString();
    }

    public static class RetryOverException
            extends IOException
    {
        public RetryOverException(String s, Throwable throwable)
        {
            super(s, throwable);
        }
    }

    public static class Config
            extends FluentdSender.Config
    {
    }
} public class SSLSender
        extends NetworkSender<SSLSocket>
{
    private final AtomicReference<SSLSocket> socket = new AtomicReference<>();
    private final SSLSocketBuilder socketBuilder;
    private final Config config;

    public SSLSender()
    {
        this(new Config());
    }

    public SSLSender(Config config)
    {
        this(config, null);
    }

    public SSLSender(FailureDetector failureDetector)
    {
        this(new Config(), failureDetector);
    }

    public SSLSender(Config config, FailureDetector failureDetector)
    {
        super(config, failureDetector);
        config.validateValues();
        socketBuilder = new SSLSocketBuilder(
                config.getHost(),
                config.getPort(),
                config.getConnectionTimeoutMilli(),
                config.getReadTimeoutMilli());
        this.config = config;
    }

    @Override
    protected SSLSocket getOrCreateSocketInternal()
            throws IOException
    {
        if (socket.get() == null) {
            socket.set(socketBuilder.build());
        }
        return socket.get();
    }

    @Override
    protected void sendBuffers(SSLSocket sslSocket, List<ByteBuffer> buffers)
            throws IOException
    {
        for (ByteBuffer buffer : buffers) {
            OutputStream outputStream = sslSocket.getOutputStream();
            if (buffer.hasArray()) {
                outputStream.write(buffer.array(), buffer.position(), buffer.remaining());
            }
            else {
                byte[] bytes = new byte[buffer.remaining()];
                buffer.get(bytes);
                outputStream.write(bytes);
            }
        }
    }

    @Override
    protected void recvResponse(SSLSocket sslSocket, ByteBuffer buffer)
            throws IOException
    {
        InputStream inputStream = sslSocket.getInputStream();
        byte[] tempBuf = new byte[buffer.remaining()];
        int read = inputStream.read(tempBuf);
        buffer.put(tempBuf, 0, read);
    }

    @Override
    protected void closeSocket()
            throws IOException
    {
        SSLSocket existingSocket;
        if ((existingSocket = socket.getAndSet(null)) != null) {
            existingSocket.close();
        }
    }

    @Override
    public String toString()
    {
        return "SSLSender{" +
                "config=" + config +
                "} " + super.toString();
    }

    public static class Config
            extends NetworkSender.Config
            implements Validatable
    {
        void validateValues()
        {
            validate();
        }
    }
} public class TCPSender
        extends NetworkSender<SocketChannel>
{
    private static final Logger LOG = LoggerFactory.getLogger(TCPSender.class);
    private final AtomicReference<SocketChannel> channel = new AtomicReference<>();
    private final Config config;

    public TCPSender()
    {
        this(new Config());
    }

    public TCPSender(Config config)
    {
        this(config, null);
    }

    public TCPSender(FailureDetector failureDetector)
    {
        this(new Config(), failureDetector);
    }

    public TCPSender(Config config, FailureDetector failureDetector)
    {
        super(config, failureDetector);
        config.validateValues();
        this.config = config;
    }

    @Override
    protected SocketChannel getOrCreateSocketInternal()
            throws IOException
    {
        if (channel.get() == null) {
            SocketChannel socketChannel = SocketChannel.open();
            try {
                socketChannel.socket().connect(new InetSocketAddress(config.getHost(), config.getPort()), config.getConnectionTimeoutMilli());
                socketChannel.socket().setTcpNoDelay(true);
                socketChannel.socket().setSoTimeout(config.getReadTimeoutMilli());
            }
            catch (Throwable e) {
                // In case of java.net.UnknownHostException and so on, the internal socket can be leak.
                // So the SocketChannel should be closed here to avoid a socket leak.
                socketChannel.close();
                throw e;
            }
            channel.set(socketChannel);
        }
        return channel.get();
    }

    @Override
    protected void sendBuffers(SocketChannel socketChannel, List<ByteBuffer> buffers)
            throws IOException
    {
        socketChannel.write(buffers.toArray(new ByteBuffer[0]));
    }

    @Override
    protected void recvResponse(SocketChannel socketChannel, ByteBuffer buffer)
            throws IOException
    {
        socketChannel.read(buffer);
    }

    @Override
    protected void closeSocket()
            throws IOException
    {
        SocketChannel existingSocketChannel;
        if ((existingSocketChannel = channel.getAndSet(null)) != null) {
            existingSocketChannel.close();
        }
    }

    @Override
    public String toString()
    {
        return "TCPSender{" +
                "config=" + config +
                "} " + super.toString();
    }

    public static class Config
            extends NetworkSender.Config
            implements Validatable
    {
        void validateValues()
        {
            validate();
        }
    }
} public class FailureDetector
        implements Heartbeater.Callback, Closeable
{
    private final FailureDetectStrategy failureDetectStrategy;
    private final Heartbeater heartbeater;
    private final AtomicReference<Long> lastFailureTimestampMillis = new AtomicReference<>();
    private final Config config;

    public FailureDetector(FailureDetectStrategy failureDetectStrategy, Heartbeater heartbeater, Config config)
    {
        config.validateValues();
        this.failureDetectStrategy = failureDetectStrategy;
        this.heartbeater = heartbeater;
        this.heartbeater.setCallback(this);
        this.heartbeater.start();
        this.config = config;
    }

    public FailureDetector(FailureDetectStrategy failureDetectStrategy, Heartbeater heartbeater)
    {
        this(failureDetectStrategy, heartbeater, new Config());
    }

    @Override
    public void onHeartbeat()
    {
        failureDetectStrategy.heartbeat(System.currentTimeMillis());
    }

    @Override
    public void onFailure(Throwable cause)
    {
        lastFailureTimestampMillis.set(System.currentTimeMillis());
    }

    public boolean isAvailable()
    {
        Long failureTimestamp = lastFailureTimestampMillis.get();
        if (failureTimestamp != null) {
            if (failureTimestamp > System.currentTimeMillis() - config.getFailureIntervalMillis()) {
                return false;
            }
            else {
                lastFailureTimestampMillis.set(null);
            }
        }
        return failureDetectStrategy.isAvailable();
    }

    @Override
    public void close()
    {
        heartbeater.close();
    }

    public FailureDetectStrategy getFailureDetectStrategy()
    {
        return failureDetectStrategy;
    }

    public Heartbeater getHeartbeater()
    {
        return heartbeater;
    }

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

    @Override
    public String toString()
    {
        return "FailureDetector{" +
                "failureDetectStrategy=" + failureDetectStrategy +
                ", heartbeater=" + heartbeater +
                ", lastFailureTimestampMillis=" + lastFailureTimestampMillis +
                ", config=" + config +
                '}';
    }

    public static class Config
        implements Validatable
    {
        @Min(0)
        private int failureIntervalMillis = 3 * 1000;

        public int getFailureIntervalMillis()
        {
            return failureIntervalMillis;
        }

        public void setFailureIntervalMillis(int failureIntervalMillis)
        {
            this.failureIntervalMillis = failureIntervalMillis;
        }

        void validateValues()
        {
            validate();
        }

        @Override
        public String toString()
        {
            return "Config{" +
                    "failureIntervalMillis=" + failureIntervalMillis +
                    '}';
        }
    }
} public class PhiAccrualFailureDetectStrategy
        extends FailureDetectStrategy
{
    private final PhiAccuralFailureDetector failureDetector;
    private final Config config;

    public PhiAccrualFailureDetectStrategy()
    {
        this(new Config());
    }

    public PhiAccrualFailureDetectStrategy(Config config)
    {
        super(config);
        config.validateValues();
        this.config = config;
        failureDetector = new PhiAccuralFailureDetector.Builder().
                setThreshold(config.getPhiThreshold()).
                setMaxSampleSize(config.getArrivalWindowSize()).
                build();
    }

    @Override
    public void heartbeat(long now)
    {
        failureDetector.heartbeat(now);
    }

    @Override
    public boolean isAvailable()
    {
        return failureDetector.isAvailable();
    }

    public double getPhiThreshold()
    {
        return config.getPhiThreshold();
    }

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

    @Override
    public String toString()
    {
        return "PhiAccrualFailureDetectStrategy{" +
                "failureDetector=" + failureDetector +
                "} " + super.toString();
    }

    public static class Config
            extends FailureDetectStrategy.Config
            implements Validatable
    {
        @DecimalMin(value = "0", inclusive = false)
        private double phiThreshold = 16;
        @Min(value = 0, inclusive = false)
        private int arrivalWindowSize = 100;

        public double getPhiThreshold()
        {
            return phiThreshold;
        }

        public void setPhiThreshold(float phiThreshold)
        {
            this.phiThreshold = phiThreshold;
        }

        public int getArrivalWindowSize()
        {
            return arrivalWindowSize;
        }

        void validateValues()
        {
            validate();
        }

        public void setArrivalWindowSize(int arrivalWindowSize)
        {
            this.arrivalWindowSize = arrivalWindowSize;
        }
    }
} public class SSLHeartbeater
        extends Heartbeater
{
    private static final Logger LOG = LoggerFactory.getLogger(SSLHeartbeater.class);
    private final Config config;
    private final SSLSocketBuilder sslSocketBuilder;

    public SSLHeartbeater()
    {
        this(new Config());
    }

    public SSLHeartbeater(final Config config)
    {
        super(config);
        config.validateValues();
        this.config = config;
        sslSocketBuilder = new SSLSocketBuilder(
                config.getHost(),
                config.getPort(),
                config.connectionTimeoutMilli,
                config.readTimeoutMilli);
    }

    @Override
    protected void invoke()
            throws IOException
    {
        try (SSLSocket sslSocket = sslSocketBuilder.build()) {
            LOG.trace("SSLHeartbeat: remotePort={}, localPort={}", sslSocket.getPort(), sslSocket.getLocalPort());
            // Try SSL handshake
            sslSocket.getSession();
            pong();
        }
    }

    @Override
    public String toString()
    {
        return "SSLHeartbeater{" +
                "config=" + config +
                ", sslSocketBuilder=" + sslSocketBuilder +
                "} " + super.toString();
    }

    public static class Config
            extends Heartbeater.Config
            implements Validatable
    {
        @Min(10)
        private int connectionTimeoutMilli = 5000;
        @Min(10)
        private int readTimeoutMilli = 5000;

        public int getConnectionTimeoutMilli()
        {
            return connectionTimeoutMilli;
        }

        public void setConnectionTimeoutMilli(int connectionTimeoutMilli)
        {
            this.connectionTimeoutMilli = connectionTimeoutMilli;
        }

        public int getReadTimeoutMilli()
        {
            return readTimeoutMilli;
        }

        public void setReadTimeoutMilli(int readTimeoutMilli)
        {
            this.readTimeoutMilli = readTimeoutMilli;
        }

        void validateValues()
        {
            validate();
        }
    }
} public class TCPHeartbeater
        extends Heartbeater
{
    private static final Logger LOG = LoggerFactory.getLogger(TCPHeartbeater.class);
    private final Config config;

    public TCPHeartbeater()
    {
        this(new Config());
    }

    public TCPHeartbeater(final Config config)
    {
        super(config);
        this.config = config;
    }

    @Override
    protected void invoke()
            throws IOException
    {
        try (SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress(config.getHost(), config.getPort()))) {
            LOG.trace("TCPHeartbeat: remotePort={}, localPort={}",
                    socketChannel.socket().getPort(), socketChannel.socket().getLocalPort());
            pong();
        }
    }

    @Override
    public String toString()
    {
        return "TCPHeartbeater{" +
                "config=" + config +
                "} " + super.toString();
    }

    public static class Config
            extends Heartbeater.Config
    {
    }
} public class ExponentialBackOffRetryStrategy
        extends RetryStrategy
{
    private final Config config;

    public ExponentialBackOffRetryStrategy()
    {
        this(new Config());
    }

    public ExponentialBackOffRetryStrategy(Config config)
    {
        super(config);
        config.validateValues();
        this.config = config;
    }

    @Override
    public int getNextIntervalMillis(int retryCount)
    {
        int interval = config.getBaseIntervalMillis() * ((int) Math.pow(2.0, (double) retryCount));
        if (interval > config.getMaxIntervalMillis()) {
            return config.getMaxIntervalMillis();
        }
        return interval;
    }

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

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

    @Override
    public String toString()
    {
        return "ExponentialBackOffRetryStrategy{" +
                "config=" + config +
                "} " + super.toString();
    }

    public static class Config
            extends RetryStrategy.Config
            implements Validatable
    {
        @Min(10)
        private int baseIntervalMillis = 400;
        @Min(10)
        private int maxIntervalMillis = 30 * 1000;

        public int getBaseIntervalMillis()
        {
            return baseIntervalMillis;
        }

        public void setBaseIntervalMillis(int baseIntervalMillis)
        {
            this.baseIntervalMillis = baseIntervalMillis;
        }

        public int getMaxIntervalMillis()
        {
            return maxIntervalMillis;
        }

        public void setMaxIntervalMillis(int maxIntervalMillis)
        {
            this.maxIntervalMillis = maxIntervalMillis;
        }

        void validateValues()
        {
            validate();
        }

        @Override
        public String toString()
        {
            return "Config{" +
                    "baseIntervalMillis=" + baseIntervalMillis +
                    ", maxIntervalMillis=" + maxIntervalMillis +
                    "} " + super.toString();
        }
    }
} public class FluentdRecordFormatter
        extends AbstractRecordFormatter
        implements RecordFormatter
{
    private static final Logger LOG = LoggerFactory.getLogger(FluentdRecordFormatter.class);

    public FluentdRecordFormatter()
    {
        this(new Config());
    }

    public FluentdRecordFormatter(Config config)
    {
        super(config);
    }

    @Override
    public byte[] format(String tag, Object timestamp, Map<String, Object> data)
    {
        try {
            return objectMapperForMessagePack.writeValueAsBytes(Arrays.asList(timestamp, data));
        }
        catch (JsonProcessingException e) {
            throw new IllegalArgumentException(
                    String.format(
                            "Failed to convert the record to MessagePack format: cause=%s, tag=%s, timestamp=%s, recordCount=%d",
                            e.getMessage(),
                            tag, timestamp, data.size())
            );
        }
    }

    @Override
    public byte[] formatFromMessagePack(String tag, Object timestamp, byte[] mapValue, int offset, int len)
    {
        try {
            ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
            // 2 items array
            outputStream.write(0x92);
            objectMapperForMessagePack.writeValue(outputStream, timestamp);
            outputStream.write(mapValue, offset, len);
            outputStream.close();

            return outputStream.toByteArray();
        }
        catch (IOException e) {
            throw new IllegalArgumentException(
                    String.format(
                            "Failed to convert the record to MessagePack format: cause=%s, tag=%s, timestamp=%s, dataSize=%s",
                            e.getMessage(),
                            tag, timestamp, mapValue.length)
            );
        }
    }

    @Override
    public byte[] formatFromMessagePack(String tag, Object timestamp, ByteBuffer mapValue)
    {
        int mapValueLen = mapValue.remaining();
        try {
            ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
            // 2 items array
            outputStream.write(0x92);
            objectMapperForMessagePack.writeValue(outputStream, timestamp);
            while (mapValue.hasRemaining()) {
                outputStream.write(mapValue.get());
            }
            outputStream.close();

            return outputStream.toByteArray();
        }
        catch (IOException e) {
            throw new IllegalArgumentException(
                    String.format(
                            "Failed to convert the record to MessagePack format: cause=%s, tag=%s, timestamp=%s, dataSize=%s",
                            e.getMessage(),
                            tag, timestamp, mapValueLen)
            );
        }
    }

    @Override
    public String formatName()
    {
        return "fluentd-packedforward";
    }

    public static class Config
            extends RecordFormatter.Config
    {
    }
} public interface Ingester
        extends Closeable
{
    void ingest(String tag, ByteBuffer dataBuffer)
            throws IOException;

    Sender getSender();
}
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;
import org.komamitsu.fluency.Fluency;
import org.komamitsu.fluency.fluentd.ingester.FluentdIngester;
import org.komamitsu.fluency.fluentd.ingester.sender.FluentdSender;
import org.komamitsu.fluency.fluentd.ingester.sender.MultiSender;
import org.komamitsu.fluency.fluentd.ingester.sender.RetryableSender;
import org.komamitsu.fluency.fluentd.ingester.sender.SSLSender;
import org.komamitsu.fluency.fluentd.ingester.sender.TCPSender;
import org.komamitsu.fluency.fluentd.ingester.sender.failuredetect.FailureDetector;
import org.komamitsu.fluency.fluentd.ingester.sender.failuredetect.PhiAccrualFailureDetectStrategy;
import org.komamitsu.fluency.fluentd.ingester.sender.heartbeat.SSLHeartbeater;
import org.komamitsu.fluency.fluentd.ingester.sender.heartbeat.TCPHeartbeater;
import org.komamitsu.fluency.fluentd.ingester.sender.retry.ExponentialBackOffRetryStrategy;
import org.komamitsu.fluency.fluentd.recordformat.FluentdRecordFormatter;
import org.komamitsu.fluency.ingester.Ingester;
        return buildFromIngester(
                recordFormatter,
                buildIngester(createBaseSender(null, port)));
    }

    public Fluency build()
    {
        return buildFromIngester(
                recordFormatter,
                buildIngester(createBaseSender(null, null)));
    }

    public Fluency build(List<InetSocketAddress> servers)
    {
        List<FluentdSender> senders = new ArrayList<>();
        for (InetSocketAddress server : servers) {
            senders.add(createBaseSender(server.getHostName(), server.getPort(), true));
        }
        return buildFromIngester(
                recordFormatter,
                buildIngester(new MultiSender(senders)));
    }

    private FluentdSender createBaseSender(String host, Integer port)
    {
        return createBaseSender(host, port, false);
    }

    private FluentdSender createBaseSender(String host, Integer port, boolean withHeartBeater)
    {
        if (withHeartBeater && port == null) {
            throw new IllegalArgumentException("`port` should be specified when using heartbeat");
        }

        if (isSslEnabled()) {
            SSLSender.Config senderConfig = new SSLSender.Config();
            FailureDetector failureDetector = null;
            if (host != null) {
                senderConfig.setHost(host);
            }
            if (port != null) {
                senderConfig.setPort(port);
            }
            if (withHeartBeater) {
                SSLHeartbeater.Config hbConfig = new SSLHeartbeater.Config();
                hbConfig.setHost(host);
                hbConfig.setPort(port);
                SSLHeartbeater heartbeater = new SSLHeartbeater(hbConfig);
                failureDetector = new FailureDetector(new PhiAccrualFailureDetectStrategy(), heartbeater);
            }
            if (connectionTimeoutMilli != null) {
                senderConfig.setConnectionTimeoutMilli(connectionTimeoutMilli);
            }
            if (readTimeoutMilli != null) {
                senderConfig.setReadTimeoutMilli(readTimeoutMilli);
            }
            return new SSLSender(senderConfig, failureDetector);
        }
        else {
            TCPSender.Config senderConfig = new TCPSender.Config();
            FailureDetector failureDetector = null;
            if (host != null) {
                senderConfig.setHost(host);
            }
            if (port != null) {
                senderConfig.setPort(port);
            }
            if (withHeartBeater) {
                TCPHeartbeater.Config hbConfig = new TCPHeartbeater.Config();
                hbConfig.setHost(host);
                hbConfig.setPort(port);
                TCPHeartbeater heartbeater = new TCPHeartbeater(hbConfig);
                failureDetector = new FailureDetector(new PhiAccrualFailureDetectStrategy(), heartbeater);
            }
            if (connectionTimeoutMilli != null) {
                senderConfig.setConnectionTimeoutMilli(connectionTimeoutMilli);
            }
            if (readTimeoutMilli != null) {
                senderConfig.setReadTimeoutMilli(readTimeoutMilli);
            }
            return new TCPSender(senderConfig, failureDetector);
        }
    }

    @Override
    public String toString()
    {
        return "FluencyBuilder{" +
                "senderMaxRetryCount=" + senderMaxRetryCount +
                ", ackResponseMode=" + ackResponseMode +
                ", sslEnabled=" + sslEnabled +
                "} " + super.toString();
    }

    private Ingester buildIngester(FluentdSender baseSender)
    {
        ExponentialBackOffRetryStrategy.Config retryStrategyConfig =
                new ExponentialBackOffRetryStrategy.Config();

        if (getSenderMaxRetryCount() != null) {
            retryStrategyConfig.setMaxRetryCount(getSenderMaxRetryCount());
        }

        if (getSenderBaseRetryIntervalMillis() != null) {
            retryStrategyConfig.setBaseIntervalMillis(getSenderBaseRetryIntervalMillis());
        }

        if (getSenderMaxRetryIntervalMillis() != null) {
            retryStrategyConfig.setMaxIntervalMillis(getSenderMaxRetryIntervalMillis());
        }

        RetryableSender.Config senderConfig = new RetryableSender.Config();

        if (getErrorHandler() != null) {
            senderConfig.setErrorHandler(getErrorHandler());
        }

        RetryableSender retryableSender = new RetryableSender(senderConfig, baseSender,
                new ExponentialBackOffRetryStrategy(retryStrategyConfig));

        FluentdIngester.Config ingesterConfig = new FluentdIngester.Config();