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 ; }
Returns a version - specific Kafka consumer with the start position configured .