private void putIntoQueue(SourceData sourceData) {
        if (sourceData == null) {
            return;
        }
        try {
            boolean offerSuc = false;
            while (isRunnable() && !offerSuc) {
                offerSuc = queue.offer(sourceData, WAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS);
            }
            if (!offerSuc) {
                MemoryManager.getInstance().release(AGENT_GLOBAL_READER_QUEUE_PERMIT, sourceData.getData().length);
            }
            LOGGER.debug("Put in source queue {} {}", new String(sourceData.getData()), inlongGroupId);
        } catch (InterruptedException e) {
            MemoryManager.getInstance().release(AGENT_GLOBAL_READER_QUEUE_PERMIT, sourceData.getData().length);
            LOGGER.error("fetchData offer failed", e);
        }
    }
After preparation work, we started to truly read data from the data source
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/file/AbstractSource.java