modified the Kafka loading mode