private void poll(long timeoutMs) {
        try {
            ConsumerRecords<K, V> records = consumer.poll(Duration.ofMillis(timeoutMs));
            for (ConsumerRecord<K, V> record : records)
                consumedCallback.onCompletion(null, record);
        } catch (WakeupException e) {
            // Expected on get() or stop(). The calling code should handle this
            throw e;
        } catch (KafkaException e) {
            log.error("Error polling: " + e);
            if (reportErrorsToCallback) {
                consumedCallback.onCompletion(e, null);
            }
        }
    }
Signals whether a topic partition should be read by this log. Invoked on {@link #start() startup} once
for every partition found in the log's backing topic.
<p>This method can be overridden by subclasses when only a subset of the assigned partitions
should be read into memory. By default, all partitions are read.
@param topicPartition A topic partition which could be read by this log.
@return true if the partition should be read by this log, false if its contents should be ignored.
connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java