JsonUtils Helper Object

JsonUtils is a Scala object with methods for serializing and deserializing Kafka TopicPartitions to and from a single JSON text.

JsonUtils uses json4s library that provides a single AST with the Jackson parser for parsing to the AST (using json4s-jackson module).

Table 1. JsonUtils API
Name	Description
partitionOffsets	Deserializing partition offsets (i.e. offsets per Kafka `TopicPartition`) from JSON, e.g. `{"topicA":{"0":23,"1":-1},"topicB":{"0":-2}}` `partitionOffsets(str: String): Map[TopicPartition, Long]`
partitionOffsets	Serializing partition offsets (i.e. offsets per Kafka `TopicPartition`) to JSON `partitionOffsets(partitionOffsets: Map[TopicPartition, Long]): String`
partitions	Deserializing `TopicPartitions` from JSON, e.g. `{"topicA":[0,1],"topicB":[0,1]}` `partitions(str: String): Array[TopicPartition]`
partitions	Serializing `TopicPartitions` to JSON `partitions(partitions: Iterable[TopicPartition]): String`

Deserializing Partition Offsets From JSON — `partitionOffsets` Method

partitionOffsets(str: String): Map[TopicPartition, Long]

partitionOffsets…FIXME

Note

partitionOffsets is used when:

KafkaSourceProvider is requested to get the desired KafkaOffsetRangeLimit (for offset option)
(Spark Structured Streaming) KafkaContinuousReader is requested to deserializeOffset
(Spark Structured Streaming) KafkaSourceOffset is created (from a SerializedOffset)

Serializing Partition Offsets to JSON — `partitionOffsets` Method

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

partitionOffsets…FIXME

Note	`partitionOffsets` is used when…FIXME

Serializing TopicPartitions to JSON — `partitions` Method

partitions(partitions: Iterable[TopicPartition]): String

partitions…FIXME

Note	`partitions` seems not to be used.

Deserializing TopicPartitions from JSON — `partitions` Method

partitions(str: String): Array[TopicPartition]

partitions uses json4s-jakson’s Serialization object to read a Map[String, Seq[Int] from the input string that represents a Map of topics and partition numbers, e.g. {"topicA":[0,1],"topicB":[0,1]}.

For every pair of topic and partition number, partitions creates a new Kafka TopicPartition.

In case of any parsing issues, partitions throws a new IllegalArgumentException:

Expected e.g. {"topicA":[0,1],"topicB":[0,1]}, got [str]

Note	`partitions` is used exclusively when `KafkaSourceProvider` is requested for a ConsumerStrategy (given assign option).

JsonUtils Helper Object