docs: @Override
  	public DataStream<Row> getDataStream(StreamExecutionEnvironment env) {
  
  		DeserializationSchema<Row> deserializationSchema = getDeserializationSchema();
  		// Version-specific Kafka consumer
  		FlinkKafkaConsumerBase<Row> kafkaConsumer = getKafkaConsumer(topic, properties, deserializationSchema);
  		return env.addSource(kafkaConsumer).name(explainSource());
  	}