Migrate a Kafka Streams transformer to a Topic View

Overview

A common pattern in event-driven architectures is to run a Kafka Streams application that reads a topic, keeps only the records a downstream team needs, removes a sensitive field, and writes the result to a second topic. It works, but it adds real operational overhead for what is, at heart, just a filter and a projection. This tutorial replaces that application with a Conduktor Gateway Topic View: a live, read-only SQL projection of a physical Kafka topic that does not copy any data. You’ll:

Run a Kafka Streams transformer that reads a global orders topic, keeps only French orders, removes the email field, and writes the result to orders_fr_kstreams.
Replace it with a Topic View orders_fr_view that expresses the same filter and projection as one SQL statement, with no application to build or run and no additional storage requirements.
Migrate the downstream consumer to the view so it picks up exactly where the transformer left off, using the correct group.id and starting offset, with no reprocessing and no data loss.

This migration simplifies operations by moving regional filtering and PII removal into the Gateway.

Prerequisites

Docker and Docker Compose (bundled with Docker Desktop).
A Conduktor license key, exported as an environment variable:
export CONDUKTOR_LICENSE_KEY=your-key-here

Clone the demo repository this tutorial follows step by step:

git clone https://github.com/conduktor/gateway-topic-views-demo
cd gateway-topic-views-demo

Stage 1 — Set up

Build and start the “before” world:

docker compose --profile before up -d --build

Before-world architecture: a producer and consumer either side of Kafka, which holds the orders and orders_fr_kstreams topics; the Kafka Streams transformer reads orders and writes orders_fr_kstreams; Conduktor Console with its Postgres database monitors Kafka

This brings up the stack shown above:

Kafka — a single broker, holding the source topic orders and the derived topic orders_fr_kstreams (three partitions each).
Kafka Streams transformer — the app being retired, compiled locally from transformer/ by --build.
Producer / consumer — the cli container. Its produce-orders.sh and consume-downstream.sh scripts stand in for the upstream producer and downstream consumer used throughout the tutorial.
Conduktor Console + Postgres — the UI and its database, used here to inspect topics and consumer groups.

Confirm both topics exist:

CLI
Console

docker compose exec cli /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --list

__consumer_offsets
orders
orders_fr_kstreams

Open Console at http://localhost:8080 and log in with admin@demo.dev / adminP4ss!, then open Topics on the backing-kafka cluster.

The Conduktor Gateway and Backing Kafka clusters listed in Conduktor Console

The orders and orders_fr_kstreams topics listed in Conduktor Console

Stage 2 — The “before” world

Orders of every country flowing through the Kafka Streams transformer, which keeps only French orders and writes them to orders_fr_kstreams for the fr-fulfillment consumer

Acting as the upstream producer, send a batch of orders:

docker compose exec cli /scripts/produce-orders.sh

This sends four orders to orders, each with an orderId key separated from the payload by |:

{"orderId":1001,"country":"FR","email":"alice@example.com","item":"Standing desk","amount":450}
{"orderId":1002,"country":"US","email":"bob@example.com","item":"Coffee mug","amount":12}
{"orderId":1003,"country":"FR","email":"chloe@example.com","item":"Office chair","amount":270}
{"orderId":1004,"country":"DE","email":"dirk@example.com","item":"Notebook","amount":6}

Acting as the downstream fr-fulfillment consumer, read the derived topic:

docker compose exec cli /scripts/consume-downstream.sh --topic orders_fr_kstreams --group fr-fulfillment

The two French orders come back, with email removed:

{"orderId":1001,"country":"FR","item":"Standing desk","amount":450}
{"orderId":1003,"country":"FR","item":"Office chair","amount":270}

See it in Console

In Console, open orders_fr_kstreams on the backing-kafka cluster and browse its messages. You will see the same two French orders, without email.

The two French orders in orders_fr_kstreams shown in Conduktor Console, with no email field

That filter-and-project is the transformer’s entire job. It reads orders, keeps only country = FR, drops email, and writes to orders_fr_kstreams. The whole service exists to run these few lines from the transformer topology:

TransformerTopology.java

// read every order
builder.stream("orders", asJson)
    // keep only French orders
    .filter((key, order) -> "FR".equals(order.get("country").asText()))
    // drop the email field
    .mapValues(order -> (JsonNode) ((ObjectNode) order).without("email"))
    // write the result
    .to("orders_fr_kstreams", toJson);

Stage 3 — Switch to the Gateway

Make sure CONDUKTOR_LICENSE_KEY is exported (the Gateway needs it), then run the switch script from the repo root:

./scripts/switch.sh

It runs the switch as four steps:

Stops the transformer.
Starts the Gateway.
Waits for the Gateway to report healthy.
Restarts Console so it re-probes the now-live Gateway.

Throughout the switch, kafka, cli, console, and postgres keep running. All Kafka data is preserved, including the orders topic and its committed offsets.

After the switch: the Kafka Streams transformer is gone and the Conduktor Gateway now proxies Kafka; Console monitors both. No Topic View exists yet and the consumer hasn't migrated

Confirm the Gateway is healthy:

CLI
Console

curl -s localhost:8888/health

{ "status": "UP", "checks": [ { "id": "live", "status": "UP" } ], "outcome": "UP" }

Open http://localhost:8080. Both clusters now appear in the cluster switcher: Conduktor Gateway and Backing Kafka. The Gateway cluster was preconfigured and — now that switch.sh has restarted Console against the live Gateway — should be reachable. Because it proxies the underlying Kafka cluster, the same topics should be visible.

The two French orders in orders_fr_kstreams shown in Conduktor Console, using the Gateway cluster

Stage 4 — Create the Topic View

curl -X PUT "http://localhost:8888/gateway/v2/topic-view" \
  -u admin:conduktor \
  -H "Content-Type: application/json" \
  --data @topic-view/orders_fr_view.json

The view definition, topic-view/orders_fr_view.json:

orders_fr_view.json

{
  "kind": "TopicView",
  "apiVersion": "gateway/v2",
  "metadata": { "name": "orders_fr_view" },
  "spec": {
    "transformation": {
      "type": "sql",
      "statement": "SELECT orderId, country, item, amount FROM orders WHERE country = 'FR'"
    },
    "onError": { "type": "DROP" }
  }
}

onError controls what the Gateway does when a record can’t be transformed (for example, non-JSON bytes the SQL can’t parse):

DROP, used here, drops the failing record and continues
FAIL_FETCH fails the fetch for that partition instead — see Error handling.

The view is now live on the Gateway as the virtual topic orders_fr_view, a read-only projection applied to the physical orders topic on every consume call.

See it in Console

Console can browse the topic view like any other topic. Switch to the Conduktor Gateway cluster and open orders_fr_view. You will see the French orders so far (1001 and 1003), with email projected away.

orders_fr_view browsed in Conduktor Console on the Gateway cluster, showing the French orders with the email field removed

You will not find the topic in the Backing Kafka cluster because it only exists as a virtual topic in Conduktor Gateway.

After-world architecture: the same producer, Kafka and consumer as the setup diagram, but the Kafka Streams transformer and the orders_fr_kstreams topic are gone — a virtual orders_fr_view on the Conduktor Gateway (shown dashed, since it stores no data) filters the physical orders topic on the fly

Stage 5 — Migrate the downstream consumer

Now switch the downstream consumer so it resumes exactly where it left off. Three rules of thumb:

Producers don’t stop for a migration — the source keeps growing, so seeding at latest skips whatever arrived in between.
The starting offset must be final — a stopped, drained pipeline gives a position that won’t move under you.
Continuity comes from the old position — starting where the old pipeline left off carries the same records forward, with no gap and no repeats.

Create backlog on the source topic

First, make the producer-side risk concrete: the switch stopped the transformer, but not the upstream producer. It keeps writing to orders while you migrate:

docker compose exec cli /scripts/produce-more-orders.sh

{"orderId":2001,"country":"FR","email":"elise@example.com","item":"Desk lamp","amount":60}
{"orderId":2002,"country":"US","email":"frank@example.com","item":"Webcam","amount":80}
{"orderId":2003,"country":"FR","email":"gaelle@example.com","item":"Monitor","amount":320}
{"orderId":2004,"country":"JP","email":"hiro@example.com","item":"Keyboard","amount":95}

With the transformer stopped, nothing consumes these four orders. They remain on orders beyond the point the transformer reached. orders-transformer is the transformer’s own consumer group, and its lag makes that visible:

docker compose exec cli /opt/kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server kafka:9092 --describe --group orders-transformer 2>/dev/null

GROUP              TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
orders-transformer orders  0          2               4               2
orders-transformer orders  1          2               2               0

CURRENT-OFFSET is where the transformer stopped reading; LOG-END-OFFSET is where orders is now. The group hasn’t advanced even though new records arrived — a running transformer would have consumed them within seconds, so the lag confirms it has stopped. Partition 0’s two unread records are part of the backlog the new consumer must pick up. Notice partition 2 isn’t listed at all: the transformer never committed an offset there, so the group has no entry for it — yet partition 2 of orders does hold unread records from the second batch. That invisible partition is the trap the migration has to handle — it must seed every partition, defaulting a never-committed one to 0, or those records would be silently skipped.

See it in Console

Console shows the same two partitions under Consumer Groups → orders-transformer → orders on the backing-kafka cluster. Like the CLI, it lists only the partitions the group has committed to, so the never-used partition 2 doesn’t appear here either.

orders-transformer consumer group lag on the orders topic shown in Conduktor Console

Confirm the old pipeline is drained

Next, make sure the old pipeline’s final position is stable. Because the transformer is stopped, it will have published no new messages to the orders_fr_kstreams topic.

CLI
Console

docker compose exec cli /opt/kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server kafka:9092 --describe --group fr-fulfillment 2>/dev/null

GROUP          TOPIC              PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
fr-fulfillment orders_fr_kstreams 0          1               1               0
fr-fulfillment orders_fr_kstreams 1          1               1               0
fr-fulfillment orders_fr_kstreams 2          0               0               0

Open Consumer Groups → fr-fulfillment on the backing-kafka cluster — zero lag across all three partitions.

fr-fulfillment consumer group with zero lag on orders_fr_kstreams shown in Conduktor Console

LAG is 0 on every partition. Combined with the stopped transformer above, that tells you the old derived-topic pipeline is fully caught up and no new records can still arrive on orders_fr_kstreams.

Understand the view’s offset model

With backlog present and the old pipeline stable, the remaining question is which offset space the view should resume from. The view resumes at a position on the input topic, orders. Two facts lead there — take them in order. 1. The old topic’s offsets don’t line up.

orders_fr_kstreams stored only the FR records, renumbered from 0 (offsets 0-3); slanting arrows show those trace back to the FR records' scattered positions on orders (offsets 1, 3, 5, 7), so the old topic's offsets don't line up with orders

The Kafka Streams app wrote only the FR records to orders_fr_kstreams, renumbered from 0. So fr-fulfillment’s committed offsets there (0–3) trace back to scattered orders offsets (1, 3, 5, 7) — they can’t tell the view where to resume. 2. The view shares the orders offset axis.

orders and orders_fr_view share one offset axis; straight vertical arrows show each orders record maps to the same offset in the view, and the filtered-out US and DE records leave gaps but the offsets carry through, so the FR records keep their orders offsets 1, 3, 5, 7

The view consumer advances over every record in orders, including the US and DE ones the filter removes — a filtered-out record still occupies an offset, so the FR records keep their orders positions with no renumbering. The resume position is therefore just an orders offset, and the group that already tracks it is the Kafka Streams app’s, orders-transformer.

Resume a view from the upstream processor’s committed offset on its input topic — never from a downstream consumer’s offsets on a separate derived topic (those are renumbered), and never from latest.

Because fr-fulfillment already drained everything the transformer produced to orders_fr_kstreams, orders-transformer’s committed position on orders is the right place for the view consumer to resume.

Name the topic view consumer group correctly

To prevent collisions between Topic Views that share the same underlying topic, the Gateway requires any consumer group that commits offsets against a Topic View to include the view name in its group.id, delimited by ::. For this tutorial we set the Topic View name as a prefix:

orders_fr_view::fr-fulfillment

Here orders_fr_view:: is the required prefix. The remainder is free-form, so this tutorial keeps the existing group name, fr-fulfillment, unchanged. Without the prefix, the Gateway rejects operations on the consumer group:

Error: Executing consumer group command failed due to Failed altering group offsets for the following partitions: [orders_fr_view-0, orders_fr_view-1, orders_fr_view-2]
org.apache.kafka.common.errors.InvalidGroupIdException: Failed altering group offsets for the following partitions: [orders_fr_view-0, orders_fr_view-1, orders_fr_view-2]

See Set the consumer group ID for the full rules, including reading from more than one view at a time.

The required group.id prefix is a temporary limitation. We plan to remove it once the Gateway supports the KIP-848 consumer group protocol.

Run the migration

With all three rules of thumb satisfied, you can seed the new consumer:

Producers don’t stop — the batch-2 backlog is sitting unprocessed on orders, waiting for the new consumer.
The starting offset must be final — the transformer is stopped and fr-fulfillment is drained, so nothing moves under the migration.
Continuity comes from the old position — orders-transformer’s committed offset on orders is where the view consumer resumes.

migrate-consumer.sh seeds a target group from a source group’s committed offsets. In this case, the source is the transformer’s group on orders, and the target is the new consumer on orders_fr_view. Run it with the required Topic View group name:

docker compose exec cli /scripts/migrate-consumer.sh \
  --source-topic orders --source-group orders-transformer \
  --target-topic orders_fr_view --target-group orders_fr_view::fr-fulfillment

GROUP                          TOPIC           PARTITION  NEW-OFFSET
orders_fr_view::fr-fulfillment orders_fr_view  1          2
orders_fr_view::fr-fulfillment orders_fr_view  0          2
orders_fr_view::fr-fulfillment orders_fr_view  2          0

The NEW-OFFSET column is the position the script set for the new group on each partition, copied straight from orders-transformer’s committed offsets on orders: partitions 0 and 1 resume at 2, where the transformer stopped, and partition 2 at 0. Partition 2 matters. The transformer never read from it during batch 1, so it had no committed offset — but the script still seeds it at 0 rather than leaving it to auto.offset.reset, so the batch-2 record waiting there isn’t skipped. See the full script, with inline comments on each decision, in scripts/migrate-consumer.sh. Inspect the migrated group — the batch-2 backlog now shows as lag on the new group:

CLI
Console

docker compose exec cli /opt/kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server gateway:6969 --describe --group orders_fr_view::fr-fulfillment 2>/dev/null

GROUP                          TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
orders_fr_view::fr-fulfillment orders          0          2               4               2
orders_fr_view::fr-fulfillment orders          1          2               2               0
orders_fr_view::fr-fulfillment orders          2          0               2               2
orders_fr_view::fr-fulfillment orders_fr_view  0          2               4               2
orders_fr_view::fr-fulfillment orders_fr_view  1          2               2               0
orders_fr_view::fr-fulfillment orders_fr_view  2          0               2               2

On the Conduktor Gateway cluster, open Consumer Groups — the migrated orders_fr_view::fr-fulfillment group appears in the list:

The Consumer Groups list on the Conduktor Gateway cluster, showing the migrated orders_fr_view::fr-fulfillment group

Open orders_fr_view::fr-fulfillment to see the topics on which the consumer group has offsets applied:

The orders_fr_view::fr-fulfillment consumer group in Conduktor Console, showing the two topics it has committed offsets on — the view orders_fr_view and its backing topic orders

Open orders_fr_view to see the seeded offsets and non-zero lag for the batch-2 records still to read:

Per-partition offsets for the orders_fr_view topic in the orders_fr_view::fr-fulfillment group, seeded at 2, 2, and 0 with non-zero lag for the pending batch-2 records

The group is positioned at 2, 2, and 0 on partitions 0, 1, and 2; the four batch-2 records beyond those offsets show as LAG, which consuming in the next stage drains.

Through the Gateway, the migrated group’s offsets are listed under both the view (orders_fr_view) and its backing orders topic. This double-listing is a known issue in Gateway 3.20.0 that will be fixed in 3.21.0.

Stage 6 — The “after” world

Orders of every country reaching the Conduktor Gateway, which applies the orders_fr_view SQL filter inline and serves only French orders to the fr-fulfillment consumer, with no transformer running

The group is migrated and the view is live. Acting as the downstream consumer again — this time reading the view — run:

docker compose exec cli /scripts/consume-downstream.sh \
  --topic orders_fr_view --group orders_fr_view::fr-fulfillment --bootstrap gateway:6969

Only the new FR orders come back: 2001 and 2003, with email removed.

{"orderId":2001,"country":"FR","item":"Desk lamp","amount":60}
{"orderId":2003,"country":"FR","item":"Monitor","amount":320}

The result shows three things:

No reprocessing — the batch-1 FR orders (1001, 1003) are not re-read. The consumer resumed at the transformer’s committed offset.
No loss — the new FR orders still came through. 2003 landed on the partition the transformer never used, which was seeded at 0; starting at latest would have skipped both new orders.
No transformer service — the same FR-only, PII-stripped output now comes from a single SQL view at read time.

Stage 7 — Clean up

docker compose --profile before --profile after down -v

Naming both profiles removes the profile-gated services, including the transformer and the Gateway. -v also drops the Kafka and Postgres volumes for a clean slate.

Conclusion

You replaced a Kafka Streams application with a single SQL Topic View and migrated its downstream consumer without reprocessing or data loss. A few things worth carrying forward:

A Topic View is a live projection, not a copy. The Gateway applies the SQL on every consume call against the physical topic, so there is no derived topic to store or keep in sync.
View consumers live in the input topic’s offset space. Migrate them by seeding the new group from the transformer’s committed offsets on the input topic, not from a downstream topic’s independent offsets.
Consumer groups targeting a Topic View need the viewName:: group prefix. It’s how the Gateway keeps two different Topic Views from overriding each other’s offsets.

Where to next:

Topic Views and logical topics — the full concept, including the SQL surface and error handling.
Set the consumer group ID — reading from one or more views.

Tutorials

Deploy

Use

Configure

Migrate

Migrate a Kafka Streams transformer to a Topic View

Overview

Prerequisites

Stage 1 — Set up

Stage 2 — The “before” world

Stage 3 — Switch to the Gateway

Stage 4 — Create the Topic View

Stage 5 — Migrate the downstream consumer

Create backlog on the source topic

Confirm the old pipeline is drained

Understand the view’s offset model

Name the topic view consumer group correctly

Run the migration

Stage 6 — The “after” world

Stage 7 — Clean up

Conclusion

​Overview

​Prerequisites

​Stage 1 — Set up

​Stage 2 — The “before” world

​Stage 3 — Switch to the Gateway

​Stage 4 — Create the Topic View

​Stage 5 — Migrate the downstream consumer

​Create backlog on the source topic

​Confirm the old pipeline is drained

​Understand the view’s offset model

​Name the topic view consumer group correctly

​Run the migration

​Stage 6 — The “after” world

​Stage 7 — Clean up

​Conclusion

Overview

Prerequisites

Stage 1 — Set up

Stage 2 — The “before” world

Stage 3 — Switch to the Gateway

Stage 4 — Create the Topic View

Stage 5 — Migrate the downstream consumer

Create backlog on the source topic

Confirm the old pipeline is drained

Understand the view’s offset model

Name the topic view consumer group correctly

Run the migration

Stage 6 — The “after” world

Stage 7 — Clean up

Conclusion