@Override
    public void destroy() {
        LOGGER.info("destroy sink {}", sourceName);
        if (!inited) {
            return;
        }
        Long start = AgentUtils.getCurrentTime();
        shutdown = true;
        LOGGER.info("destroy proxySink, wait for sender close {} ms instance {}", AgentUtils.getCurrentTime() - start,
                profile.getInstanceId());
        start = AgentUtils.getCurrentTime();
        while (running || offsetRunning) {
            AgentUtils.silenceSleepInMs(LOOP_WAIT_TIME_MS);
        }
        LOGGER.info("destroy proxySink, wait for run close {} ms instance {}", AgentUtils.getCurrentTime() - start,
                profile.getInstanceId());
        start = AgentUtils.getCurrentTime();
        clearOffset();
        LOGGER.info("destroy proxySink, wait for offset clear {} ms instance {}", AgentUtils.getCurrentTime() - start,
                profile.getInstanceId());
        LOGGER.info("destroy sink {} end", sourceName);
    }
flush cache by batch

@return thread runner
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/ProxySink.java