code: protected FlinkKafkaConsumerBase<Row> getKafkaConsumer(
  			String topic,
  			Properties properties,
  			DeserializationSchema<Row> deserializationSchema) {
  		FlinkKafkaConsumerBase<Row> kafkaConsumer =
  				createKafkaConsumer(topic, properties, deserializationSchema);
  		switch (startupMode) {
  			case EARLIEST:
  				kafkaConsumer.setStartFromEarliest();
  				break;
  			case LATEST:
  				kafkaConsumer.setStartFromLatest();
  				break;
  			case GROUP_OFFSETS:
  				kafkaConsumer.setStartFromGroupOffsets();
  				break;
  			case SPECIFIC_OFFSETS:
  				kafkaConsumer.setStartFromSpecificOffsets(specificStartupOffsets);
  				break;
  		}
  		return kafkaConsumer;
  	}
docs: Returns a version-specific Kafka consumer with the start position configured.
  
  @param topic                 Kafka topic to consume.
  @param properties            Properties for the Kafka consumer.
  @param deserializationSchema Deserialization schema to use for Kafka records.
  @return The version-specific Kafka consumer