spark-sql-streaming-KafkaOffsetReader.adoc

KafkaOffsetReader

KafkaOffsetReader is created when:

• KafkaRelation builds an RDD with rows that are records from Kafka

• KafkaSourceProvider creates a KafkaSource (for kafka format)

Table 1. KafkaOffsetReader’s Options

Name	Default Value	Description
fetchOffset.numRetries	3
fetchOffset.retryIntervalMs	1000	How long to wait before retries.

KafkaOffsetReader defines the predefined fixed schema of Kafka source.

Table 2. KafkaOffsetReader’s Internal Registries and Counters

Name	Description
consumer	Kafka’s Consumer (with keys and values of Array[Byte] type) Initialized when KafkaOffsetReader is created. Used when KafkaOffsetReader: fetchTopicPartitions fetches offsets for selected TopicPartitions fetchEarliestOffsets fetchLatestOffsets resetConsumer is closed
execContext
groupId
kafkaReaderThread
maxOffsetFetchAttempts
nextId
offsetFetchAttemptIntervalMs

Tip	Enable INFO or DEBUG logging levels for org.apache.spark.sql.kafka010.KafkaOffsetReader to see what happens inside. Add the following line to conf/log4j.properties: log4j.logger.org.apache.spark.sql.kafka010.KafkaOffsetReader=DEBUG Refer to Logging.

nextGroupId Internal Method

Caution

FIXME

resetConsumer Internal Method

Caution

FIXME

fetchTopicPartitions Method

fetchTopicPartitions(): Set[TopicPartition]

Caution

FIXME

Note	fetchTopicPartitions is used when KafkaRelation getPartitionOffsets.

Fetching Earliest Offsets — fetchEarliestOffsets Method

fetchEarliestOffsets(newPartitions: Seq[TopicPartition]): Map[TopicPartition, Long]

Caution

FIXME

Note	fetchEarliestOffsets is used when KafkaSource rateLimit and generates a DataFrame for a batch (when new partitions have been assigned).

Fetching Latest Offsets — fetchLatestOffsets Method

fetchLatestOffsets(): Map[TopicPartition, Long]

Caution

FIXME

Note	fetchLatestOffsets is used when KafkaSource gets offsets or initialPartitionOffsets is initialized.

withRetriesWithoutInterrupt Internal Method

withRetriesWithoutInterrupt(body: => Map[TopicPartition, Long]): Map[TopicPartition, Long]

Creating KafkaOffsetReader Instance

KafkaOffsetReader takes the following when created:

• ConsumerStrategy

• Kafka parameters (as name-value pairs that are used exclusively to create a Kafka consumer

• Options (as name-value pairs)

• Prefix for group id

KafkaOffsetReader initializes the internal registries and counters.

Fetching Offsets for Selected TopicPartitions — fetchSpecificOffsets Method

fetchSpecificOffsets(partitionOffsets: Map[TopicPartition, Long]): Map[TopicPartition, Long]

Figure 1. KafkaOffsetReader’s fetchSpecificOffsets

fetchSpecificOffsets requests the Kafka Consumer to poll(0).

fetchSpecificOffsets requests the Kafka Consumer for assigned partitions (using Consumer.assignment()).

fetchSpecificOffsets requests the Kafka Consumer to pause(partitions).

You should see the following DEBUG message in the logs:

DEBUG KafkaOffsetReader: Partitions assigned to consumer: [partitions]. Seeking to [partitionOffsets]

For every partition offset in the input partitionOffsets, fetchSpecificOffsets requests the Kafka Consumer to:

• seekToEnd for the latest (aka -1)

• seekToBeginning for the earliest (aka -2)

• seek for other offsets

In the end, fetchSpecificOffsets creates a collection of Kafka’s TopicPartition and position (using the Kafka Consumer).

Note	fetchSpecificOffsets is used when KafkaSource fetches and verifies initial partition offsets.

Creating Kafka Consumer — createConsumer Internal Method

createConsumer(): Consumer[Array[Byte], Array[Byte]]

createConsumer requests ConsumerStrategy to create a Kafka Consumer with driverKafkaParams and new generated group.id Kafka property.

Note	createConsumer is used when KafkaOffsetReader is created (and initializes consumer) and resetConsumer