becketqin commented on a change in pull request #17991:
URL: https://github.com/apache/flink/pull/17991#discussion_r769321825
##########
File path:
flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java
##########
@@ -105,84 +91,38 @@ public KafkaPartitionSplitReader(
}
@Override
- public RecordsWithSplitIds<Tuple3<T, Long, Long>> fetch() throws
IOException {
- KafkaPartitionSplitRecords<Tuple3<T, Long, Long>> recordsBySplits =
- new KafkaPartitionSplitRecords<>();
+ public RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>> fetch() throws
IOException {
ConsumerRecords<byte[], byte[]> consumerRecords;
try {
consumerRecords = consumer.poll(Duration.ofMillis(POLL_TIMEOUT));
} catch (WakeupException we) {
- recordsBySplits.prepareForRead();
- return recordsBySplits;
+ return null;
Review comment:
It seems that rerunning null here still might cause NPE in the
`FetchTask.run()` even though null is only returned when the consumer is
wakenup.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
