private void connectWithRetry(boolean forceConnect) {
        if (!isConnected.get()) {
            if (!isTryingToConnect.getAndSet(true) || forceConnect) {
                try {
                    connect();
                    setConnected(true);
                    isTryingToConnect.set(false);
                    if (connectionCallback != null) {
                        connectionCallback.connectionEstablished();
                    }
                    backoffRetryCounter.reset();
                } catch (ConnectionUnavailableException e) {
                    LOG.error(StringUtil.removeCRLFCharacters(
                            ExceptionUtil.getMessageWithContext(e, siddhiAppContext) +
                                    ", error while connecting at Sink '" + type + "' at '" + streamDefinition.getId() +
                                    "', will retry in '" + backoffRetryCounter.getTimeInterval() + "'."), e);
                    scheduledExecutorService.schedule(new Runnable() {
                        @Override
                        public void run() {
                            connectWithRetry(true);
                        }
                    }, backoffRetryCounter.getTimeIntervalMillis(), TimeUnit.MILLISECONDS);
                    backoffRetryCounter.increment();
                } catch (RuntimeException e) {
                    LOG.error(StringUtil.removeCRLFCharacters(
                            ExceptionUtil.getMessageWithContext(e, siddhiAppContext)) +
                            ", error while connecting at Sink '" + StringUtil.removeCRLFCharacters(type) + "' at '" +
                            StringUtil.removeCRLFCharacters(streamDefinition.getId()) + "'.", e);
                    throw e;
                }
            }
        }
    }
Called at the end to clean all the resources consumed
modules/siddhi-core/src/main/java/io/siddhi/core/stream/output/sink/Sink.java