[FLINK-9627] Extending 'KafkaJsonTableSource' according to comments will result in NPE

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.5.0
• Fix Version/s: 1.4.3, 1.5.1, 1.6.0
• Labels: pull-request-available
• Assignee: vinoyang
• Reporter: Dominik Wosiński
• Created: 20/Jun/18 13:17
• Updated: 29/Jun/18 16:54
• Resolved: 29/Jun/18 16:54

Description

According to the comments what is needed to extend the 'KafkaJsonTableSource' looks as follows:

 

A version-agnostic Kafka JSON {@link StreamTableSource}.
*
* <p>The version-specific Kafka consumers need to extend this class and
* override {@link #createKafkaConsumer(String, Properties, DeserializationSchema)}}.
*
* <p>The field names are used to parse the JSON file and so are the types.

This will cause an NPE, since there is no default value for startupMode in the abstract class itself only in the builder of this class.
For the 'getKafkaConsumer' method the switch statement will be executed on non-initialized 'startupMode' field:

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;
}

 

 

Issue Links

• links to: GitHub Pull Request #6198