private boolean putInCache(Message message) {
        try {
            if (message == null) {
                return true;
            }
            extractStreamFromMessage(message, fieldSplitter);
            if (message instanceof EndMessage) {
                // increment the count of failed sinks
                sinkMetric.sinkFailCount.incrementAndGet();
                return true;
            }
            ProxyMessage proxyMessage = new ProxyMessage(message);
            boolean writerPermitSuc = MemoryManager.getInstance()
                    .tryAcquire(AGENT_GLOBAL_WRITER_PERMIT, message.getBody().length);
            if (!writerPermitSuc) {
                MemoryManager.getInstance().printDetail(AGENT_GLOBAL_WRITER_PERMIT, "proxy sink");
                return false;
            }
            // add message to package proxy
            boolean suc = cache.add(proxyMessage);
            if (suc) {
                addAckInfo(proxyMessage.getAckInfo());
            } else {
                MemoryManager.getInstance().release(AGENT_GLOBAL_WRITER_PERMIT, message.getBody().length);
                // increment the count of failed sinks
                sinkMetric.sinkFailCount.incrementAndGet();
            }
            return suc;
        } catch (Exception e) {
            LOGGER.error("write message to Proxy sink error", e);
        } catch (Throwable t) {
            ThreadUtils.threadThrowableHandler(Thread.currentThread(), t);
        }
        return false;
    }
sink message data to inlong-dataproxy
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/ProxySink.java