@Override
    protected void initSource(InstanceProfile profile) {
        try {
            LOGGER.info("KafkaSource init: {}", profile.toJsonStr());
            topic = profile.getInstanceId();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, profile.get(TASK_KAFKA_BOOTSTRAP_SERVERS));
            props.put(ConsumerConfig.GROUP_ID_CONFIG, taskId);
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, KAFKA_DESERIALIZER_METHOD);
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KAFKA_DESERIALIZER_METHOD);
            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, profile.get(TASK_KAFKA_AUTO_COMMIT_OFFSET_RESET));
            props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

            allPartitionOffsets = profile.get(TASK_KAFKA_OFFSET, null);
            isRestoreFromDB = profile.getBoolean(RESTORE_FROM_DB, false);
            if (!isRestoreFromDB && StringUtils.isNotBlank(allPartitionOffsets)) {
                // example:0#110_1#666_2#222
                String[] offsets = allPartitionOffsets.split(TASK_KAFKA_OFFSET_DELIMITER);
                for (String offset : offsets) {
                    partitionOffsets.put(Integer.valueOf(offset.split(TASK_KAFKA_PARTITION_OFFSET_DELIMITER)[0]),
                            Long.valueOf(offset.split(TASK_KAFKA_PARTITION_OFFSET_DELIMITER)[1]));
                }
            }
            kafkaConsumer = getKafkaConsumer();
        } catch (Exception ex) {
            stopRunning();
            throw new FileException("error init stream for " + topic, ex);
        }
    }
kafka source, split kafka source job into multi readers
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/KafkaSource.java