@Override
    public void init(InstanceProfile profile) {
        super.init(profile);
        fieldSplitter = profile.get(CommonConstants.FIELD_SPLITTER, DEFAULT_FIELD_SPLITTER).getBytes(
                StandardCharsets.UTF_8);
        sourceName = profile.getInstanceId();
        offsetManager = OffsetManager.getInstance();
        sender = SenderManager.getInstance().getSender(profile.getTaskId(), profile);
        try {
            EXECUTOR_SERVICE.execute(coreThread());
            EXECUTOR_SERVICE.execute(flushOffset());
            inited = true;
        } catch (Throwable ex) {
            shutdown = true;
            LOGGER.error("error while init sender for group id {}", inlongGroupId);
            ThreadUtils.threadThrowableHandler(Thread.currentThread(), ex);
            throw new IllegalStateException(ex);
        }
    }
flush cache by batch

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