public void sendMessageFromCache() {
        ConcurrentHashMap<String, LinkedBlockingQueue<ProxyMessage>> messageQueueMap = cache.getMessageQueueMap();
        for (Map.Entry<String, LinkedBlockingQueue<ProxyMessage>> entry : messageQueueMap.entrySet()) {
            SenderMessage senderMessage = cache.fetchSenderMessage(entry.getKey(), entry.getValue());
            if (senderMessage == null) {
                continue;
            }
            sender.sendBatch(senderMessage);
            if (AgentUtils.getCurrentTime() - lastPrintTime > TimeUnit.SECONDS.toMillis(1)) {
                lastPrintTime = AgentUtils.getCurrentTime();
                LOGGER.info("send groupId {}, streamId {}, message size {}, taskId {}, "
                        + "instanceId {} sendTime is {}", inlongGroupId, inlongStreamId,
                        senderMessage.getDataList().size(), profile.getTaskId(),
                        profile.getInstanceId(),
                        senderMessage.getDataTime());
            }
        }
    }
flush cache by batch

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