> ## Documentation Index
> Fetch the complete documentation index at: https://docs.conduktor.io/llms.txt
> Use this file to discover all available pages before exploring further.
# Migrate a Kafka Streams transformer to a Topic View
> Replace a Kafka Streams filter-and-project app with a Conduktor Gateway Topic View, and migrate the downstream consumer's group.id and offset without data loss.
## 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](/guide/conduktor-concepts/logical-topics#topic-views): a live, read-only SQL projection of a physical Kafka topic that does not copy any data.
You'll:
1. 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`.
2. 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.
3. 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](https://docs.docker.com/get-docker/) and [Docker Compose](https://docs.docker.com/compose/install/) (bundled with Docker Desktop).
* A Conduktor license key, exported as an environment variable:
```bash theme={null}
export CONDUKTOR_LICENSE_KEY=your-key-here
```
* Clone the demo repository this tutorial follows step by step:
```bash theme={null}
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:
```bash theme={null}
docker compose --profile before up -d --build
```
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:
```bash theme={null}
docker compose exec cli /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --list
```
```text theme={null}
__consumer_offsets
orders
orders_fr_kstreams
```
Open Console at [http://localhost:8080](http://localhost:8080) and log in with `admin@demo.dev` / `adminP4ss!`, then open **Topics** on the `backing-kafka` cluster.
## Stage 2 — The "before" world
Acting as the upstream producer, send a batch of orders:
```bash theme={null}
docker compose exec cli /scripts/produce-orders.sh
```
This sends four orders to `orders`, each with an `orderId` key separated from the payload by `|`:
```text theme={null}
1001|{"orderId":1001,"country":"FR","email":"alice@example.com","item":"Standing desk","amount":450}
1002|{"orderId":1002,"country":"US","email":"bob@example.com","item":"Coffee mug","amount":12}
1003|{"orderId":1003,"country":"FR","email":"chloe@example.com","item":"Office chair","amount":270}
1004|{"orderId":1004,"country":"DE","email":"dirk@example.com","item":"Notebook","amount":6}
```
Acting as the downstream `fr-fulfillment` consumer, read the derived topic:
```bash theme={null}
docker compose exec cli /scripts/consume-downstream.sh --topic orders_fr_kstreams --group fr-fulfillment
```
The two French orders come back, with `email` removed:
```json theme={null}
{"orderId":1001,"country":"FR","item":"Standing desk","amount":450}
{"orderId":1003,"country":"FR","item":"Office chair","amount":270}
```
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`.
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](https://github.com/conduktor/gateway-topic-views-demo/blob/main/transformer/src/main/java/io/conduktor/demo/TransformerTopology.java):
```java title="TransformerTopology.java" theme={null}
// 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:
```bash theme={null}
./scripts/switch.sh
```
It runs the switch as four steps:
1. Stops the transformer.
2. Starts the Gateway.
3. Waits for the Gateway to report healthy.
4. 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.
Confirm the Gateway is healthy:
```bash theme={null}
curl -s localhost:8888/health
```
```json theme={null}
{ "status": "UP", "checks": [ { "id": "live", "status": "UP" } ], "outcome": "UP" }
```
Open [http://localhost:8080](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.
## Stage 4 — Create the Topic View
Register the view with a single call to the Gateway's admin API:
```bash theme={null}
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`:
```json title="orders_fr_view.json" theme={null}
{
"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](/guide/conduktor-concepts/logical-topics#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.
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.
You will not find the topic in the **Backing Kafka** cluster because it only exists as a virtual topic in **Conduktor Gateway**.
## 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:
```bash theme={null}
docker compose exec cli /scripts/produce-more-orders.sh
```
```text theme={null}
2001|{"orderId":2001,"country":"FR","email":"elise@example.com","item":"Desk lamp","amount":60}
2002|{"orderId":2002,"country":"US","email":"frank@example.com","item":"Webcam","amount":80}
2003|{"orderId":2003,"country":"FR","email":"gaelle@example.com","item":"Monitor","amount":320}
2004|{"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:
```bash theme={null}
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.
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.
### 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.
```bash theme={null}
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.
`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.**
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.**
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](/guide/conduktor-concepts/logical-topics#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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-848%3A+The+Next+Generation+of+the+Consumer+Rebalance+Protocol) 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:
```bash theme={null}
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`](https://github.com/conduktor/gateway-topic-views-demo/blob/main/scripts/migrate-consumer.sh).
Inspect the migrated group — the batch-2 backlog now shows as lag on the new group:
```bash theme={null}
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:
Open **orders\_fr\_view::fr-fulfillment** to see the topics on which the consumer group has offsets applied:
Open **orders\_fr\_view** to see the seeded offsets and non-zero lag for the batch-2 records still to read:
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
The group is migrated and the view is live. Acting as the downstream consumer again — this time reading the view — run:
```bash theme={null}
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.
```json theme={null}
{"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
```bash theme={null}
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](/guide/conduktor-concepts/logical-topics#topic-views) — the full concept, including the SQL surface and error handling.
* [Set the consumer group ID](/guide/conduktor-concepts/logical-topics#set-the-consumer-group-id) — reading from one or more views.