How to update the number of partitions of a Kafka topic with ksqlDB

Imagine you want to change the partitions of your Kafka topic. You can use a streaming transformation to automatically stream all the messages from the original topic into a new Kafka topic that has the desired number of partitions.

Setup

To accomplish this transformation, first create a stream based on the original topic:

CREATE STREAM s1 (k VARCHAR KEY, v VARCHAR)
    WITH (KAFKA_TOPIC='topic',
          VALUE_FORMAT='JSON');

Then, create a second stream that reads everything from the original topic and puts into a new topic with the desired number of partitions:

CREATE STREAM s2
    WITH (KAFKA_TOPIC='topic2',
          VALUE_FORMAT='JSON',
          PARTITIONS=2) AS
    SELECT *
    FROM s1
    EMIT CHANGES;

Running the example

Prerequisites

• Docker running via Docker Desktop or Docker Engine
• Docker Compose. Ensure that the command docker compose version succeeds.

Run the commands

Clone the confluentinc/tutorials GitHub repository (if you haven't already) and navigate to the tutorials directory:

git clone git@github.com:confluentinc/tutorials.git
cd tutorials

Start ksqlDB and Kafka:

docker compose -f ./docker/docker-compose-ksqldb.yml up -d

Next, open the ksqlDB CLI:

docker exec -it ksqldb-cli ksql http://ksqldb-server:8088

Run the following SQL statements to create the s1 stream backed by Kafka running in Docker and populate it with test data.

CREATE STREAM s1 (k VARCHAR KEY, v VARCHAR)
    WITH (KAFKA_TOPIC='topic1',
          PARTITIONS=1,
          VALUE_FORMAT='JSON');

INSERT INTO s1 (k, v) VALUES ('hello', 'world');
INSERT INTO s1 (k, v) VALUES ('foo', 'bar');
INSERT INTO s1 (k, v) VALUES ('bar', 'baz');

Next, run the CREATE STREAM AS SELECT query to populate a new topic, topic2 with the same events in topic1 but having 2 partitions.

SET 'auto.offset.reset'='earliest';

CREATE STREAM s2
    WITH (KAFKA_TOPIC='topic2',
          VALUE_FORMAT='JSON',
          PARTITIONS=2) AS
    SELECT *
    FROM s1
    EMIT CHANGES;

Observe the expected number of partitions when you run the kafka-topics command in the broker container:

docker exec -it broker kafka-topics --bootstrap-server localhost:29092 --describe --topic topic1

docker exec -it broker kafka-topics --bootstrap-server localhost:29092 --describe --topic topic2

When you are finished, exit the ksqlDB CLI by entering CTRL-D and clean up the containers used for this tutorial by running:

docker compose -f ./docker/docker-compose-ksqldb.yml down