@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());
	}
@ Override public DataStream < Row > getDataStream ( StreamExecutionEnvironment env ) { DeserializationSchema < Row > deserializationSchema = getDeserializationSchema ( ) ; FlinkKafkaConsumerBase < Row > kafkaConsumer = getKafkaConsumer ( topic , properties , deserializationSchema ) ; return env . addSource ( kafkaConsumer ) . name ( explainSource ( ) ) ; }
NOTE: This method is for internal use only for defining a TableSource.
Do not use it in Table API programs.