private KafkaConsumer<String, byte[]> getKafkaConsumer() {
        List<PartitionInfo> partitionInfoList;
        KafkaConsumer<String, byte[]> kafkaConsumer = null;
        props.put(KAFKA_SESSION_TIMEOUT, 30000);
        try {
            kafkaConsumer = new KafkaConsumer<>(props);
            partitionInfoList = kafkaConsumer.partitionsFor(topic);
            if (partitionInfoList == null) {
                kafkaConsumer.close();
                return null;
            }
            List<TopicPartition> topicPartitions = new ArrayList<>();
            for (PartitionInfo partitionInfo : partitionInfoList) {
                TopicPartition topicPartition = new TopicPartition(partitionInfo.topic(),
                        partitionInfo.partition());
                topicPartitions.add(topicPartition);
            }
            kafkaConsumer.assign(topicPartitions);
            if (!isRestoreFromDB && StringUtils.isNotBlank(allPartitionOffsets)) {
                for (TopicPartition topicPartition : topicPartitions) {
                    Long offset = partitionOffsets.get(topicPartition.partition());
                    if (ObjectUtils.isNotEmpty(offset)) {
                        kafkaConsumer.seek(topicPartition, offset);
                    }
                }
            } else {
                LOGGER.info("Skip to seek offset");
            }
            return kafkaConsumer;
        } catch (Exception e) {
            if (kafkaConsumer != null) {
                kafkaConsumer.close();
            }
            LOGGER.error("get kafka consumer error", e);
        }
        return null;
    }
kafka source, split kafka source job into multi readers
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/KafkaSource.java