@Override
    protected List<SourceData> readFromSource() {
        List<SourceData> dataList = new ArrayList<>();
        ConsumerRecords<String, byte[]> records = kafkaConsumer.poll(Duration.ofMillis(1000));
        for (ConsumerRecord<String, byte[]> record : records) {
            SourceData sourceData = new SourceData(record.value(), Long.toString(record.offset()));
            dataList.add(sourceData);
            offset = record.offset();
        }
        kafkaConsumer.commitSync();
        return dataList;
    }
kafka source, split kafka source job into multi readers
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/KafkaSource.java