public void onError(Object payload, DynamicOptions dynamicOptions, Exception e) {
        OnErrorAction errorAction = onErrorAction;
        if (e instanceof ConnectionUnavailableException) {
            setConnected(false);
            if (connectionCallback != null) {
                connectionCallback.connectionFailed();
            }
        } else if (errorAction == OnErrorAction.WAIT) {
            LOG.error("Error on '" + siddhiAppContext.getName() + "'. Dropping event at Sink '"
                    + type + "' at '" + streamDefinition.getId() + "' as on.error='wait' does not handle " +
                    "'" + e.getClass().getName() + "' error: '" + e.getMessage() + "', events dropped '" +
                    payload + "'", e);
            return;
        }
        try {
            switch (errorAction) {
                case STREAM:
                    connectWithRetry();
                    streamJunction.handleError(dynamicOptions.getEvent(), e);
                    break;
                case WAIT:
                    LOG.error(StringUtil.removeCRLFCharacters(
                            ExceptionUtil.getMessageWithContext(e, siddhiAppContext) +
                                    ", error while connecting Sink '" + type + "' at '" + streamDefinition.getId() +
                                    "', will retry every '5 sec'."), e);
                    int count = 0;
                    while (!isConnected.get()) {
                        if (isShutdown.get()) {
                            isTryingToConnect.set(false);
                            return;
                        }
                        retryWait(5000);
                        count++;
                        if (!isConnected.get() && !isTryingToConnect.getAndSet(true)) {
                            while (!isConnected.get()) {
                                if (isShutdown.get()) {
                                    isTryingToConnect.set(false);
                                    return;
                                }
                                S state = stateHolder.getState();
                                try {
                                    connectAndPublish(payload, dynamicOptions, state);
                                    isTryingToConnect.set(false);
                                    return;
                                } catch (ConnectionUnavailableException ignore) {
                                } finally {
                                    stateHolder.returnState(state);
                                }
                                if (count % 12 == 0) {
                                    LOG.warn(StringUtil.removeCRLFCharacters(
                                            ExceptionUtil.getMessageWithContext(e, siddhiAppContext) +
                                                    ", still waiting to connect Sink '" + type +
                                                    "' at '" + streamDefinition.getId() +
                                                    "' retrying every '5 sec'."), e);
                                }
                                retryWait(5000);
                                count++;
                            }
                        }
                        if (count % 12 == 0) {
                            LOG.warn(StringUtil.removeCRLFCharacters(
                                    ExceptionUtil.getMessageWithContext(e, siddhiAppContext) +
                                            ", still waiting to connect Sink '" + type +
                                            "' at '" + streamDefinition.getId() + "."), e);
                        }
                    }
                    S state = stateHolder.getState();
                    try {
                        publish(payload, dynamicOptions, state);
                    } catch (ConnectionUnavailableException ignore) {
                        onError(payload, dynamicOptions, e);
                    } finally {
                        stateHolder.returnState(state);
                    }
                    break;
                case STORE:
                    connectWithRetry();
                    ErroneousEvent erroneousEvent = new ErroneousEvent(dynamicOptions.getEvent(), e, e.getMessage());
                    erroneousEvent.setOriginalPayload(payload);
                    ErrorStoreHelper.storeErroneousEvent(siddhiAppContext.getSiddhiContext().getErrorStore(),
                            ErrorOccurrence.STORE_ON_SINK_ERROR, siddhiAppContext.getName(),
                            erroneousEvent, streamDefinition.getId());
                    break;
                case LOG:
                default:
                    connectWithRetry();
                    LOG.error("Error on '" + siddhiAppContext.getName() + "'. Dropping event at Sink '"
                            + type + "' at '" + streamDefinition.getId() + "' as its still trying to reconnect!, "
                            + "events dropped '" + payload + "'");
                    if (LOG.isDebugEnabled()) {
                        LOG.debug(e);
                    }
                    break;
            }
        } catch (Throwable t) {
            LOG.error("Error on '" + siddhiAppContext.getName() + "'. Dropping event at Sink '"
                    + type + "' at '" + streamDefinition.getId() + "' as there is an issue when " +
                    "handling the error: '" + t.getMessage() + "', events dropped '" + payload + "'", 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