Partitions can be added to a Kafka topic dynamically. This example shows that an existing stream will not see the data published to the new partitions, and only when the existing streaming context is terminated and a new stream is started from a new context will that data be delivered.

The topic is created with three partitions, and so each RDD the stream produces has three partitions as well, even after two more partitions are added to the topic. This is what's received after the first 500 records are published to the topic while it has only three partitions:

[1] *** got an RDD, size = 500
[1] *** 3 partitions
[1] *** partition size = 155
[1] *** partition size = 173
[1] *** partition size = 172

When two partitions are added and another 500 messages are published, this is what's received (note both the number of partitions and the number of messages):

[1] *** got an RDD, size = 288
[1] *** 3 partitions
[1] *** partition size = 98
[1] *** partition size = 89
[1] *** partition size = 101

When a new stream is subsequently created, the RDDs produced have five partitions, but only two of them contain data, as all the data has been drained from the initial three partitions of the topic, by the first stream. Now all 500 messages (288 + 212) from the second set have been delivered.

[2] *** got an RDD, size = 212
[2] *** 5 partitions
[2] *** partition size = 0
[2] *** partition size = 0
[2] *** partition size = 0
[2] *** partition size = 112
[2] *** partition size = 100

• The topics have three and six partitions respectively.
• Each RDD has nine partitions.
• Each RDD partition receives data from exactly one partition of one topic.

*** got an RDD, size = 200
*** 9 partitions
*** partition 1 has 27 records
*** rdd partition = 1, topic = foo, topic partition = 0, record count = 27.
*** partition 2 has 15 records
*** rdd partition = 2, topic = bar, topic partition = 1, record count = 15.
*** partition 3 has 17 records
*** rdd partition = 3, topic = bar, topic partition = 0, record count = 17.
*** partition 4 has 39 records
*** rdd partition = 4, topic = foo, topic partition = 1, record count = 39.
*** partition 5 has 34 records
*** rdd partition = 5, topic = foo, topic partition = 2, record count = 34.
*** partition 6 has 11 records
*** rdd partition = 6, topic = bar, topic partition = 3, record count = 11.
*** partition 7 has 18 records
*** rdd partition = 7, topic = bar, topic partition = 4, record count = 18.
*** partition 8 has 20 records
*** rdd partition = 8, topic = bar, topic partition = 2, record count = 20.

Record timestamps were introduced into Kafka 0.10 as described in KIP-32 and KIP-33.

This example sets up two different topics that handle timestamps differently -- topic A has the timestamp set by the broker when it receives the record, while topic B passes through the timestamp provided in the record (either programmatically when the record was created, as shown here, or otherwise automatically by the producer.)

Since the record carries information about where its timestamp originates, its easy to subscribe to the two topics to create a single stream, and then examine the timestamp of every received record and its type.

NOTE: The use of timestamps to filter topics in the broker, as introduced in Kafka 0.10.1, is blocked on SPARK-18057.