# Compacted Topics (/connectors/kafka/how-to/compacted-topics)



Log compaction keeps only the **latest record per key** instead of aging a topic out by time. This
page walks through turning it on — at topic creation or on an existing topic — producing keyed
records and a tombstone, and what the background compactor does with them. For the durability and
retention model compaction sits alongside, see
[Durability & Retention](/connectors/kafka/concepts/durability-and-retention); this page is
the practical companion.

<Callout type="info">
  **Runtime support, not a migration path.** Turning on `cleanup.policy=compact` is a live operation
  on a topic that already lives on KubeMQ — nothing here moves data. Bringing an **existing**
  compacted topic's history over from a real Kafka cluster is a separate, start-fresh adoption story
  with its own assess/replicate/cutover playbook — see
  [Migrate from Kafka](/connectors/kafka/how-to/migrate-from-kafka) rather than treating
  compaction as something you "migrate."
</Callout>

## Create a compacted topic [#create-a-compacted-topic]

Set `cleanup.policy=compact` in the topic's config at `CreateTopics` time — the natural path for a
new topic, and the one every Kafka Connect internal topic and Kafka Streams changelog topic already
uses. `kcat` has no topic-admin API (it only produces and consumes), so the example below uses each
client library's admin surface instead; produce and consume with whichever client you like once the
topic exists.

<Tabs groupId="language" items="['Go', 'Python', 'Java', 'JavaScript', 'C#', 'Ruby', 'Rust']">
  <Tab value="Go">
    ```go
    package main

    import (
    	"context"
    	"log"

    	"github.com/twmb/franz-go/pkg/kadm"
    	"github.com/twmb/franz-go/pkg/kgo"
    )

    func main() {
    	ctx := context.Background()

    	cl, err := kgo.NewClient(kgo.SeedBrokers("localhost:9092"))
    	if err != nil {
    		log.Fatalf("client: %v", err)
    	}
    	defer cl.Close()

    	admin := kadm.NewClient(cl)
    	resp, err := admin.CreateTopic(ctx, 1, 1, map[string]*string{
    		"cleanup.policy": kadm.StringPtr("compact"),
    	}, "user-profiles")
    	if err != nil || resp.Err != nil {
    		log.Fatalf("create topic: %v / %v", err, resp.Err)
    	}
    	log.Println("created user-profiles with cleanup.policy=compact")
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python
    from confluent_kafka.admin import AdminClient, NewTopic

    admin = AdminClient({"bootstrap.servers": "localhost:9092"})
    topic = NewTopic("user-profiles", num_partitions=1, replication_factor=1,
                      config={"cleanup.policy": "compact"})

    futures = admin.create_topics([topic])
    for name, future in futures.items():
        future.result()  # raises on failure
        print(f"created {name} with cleanup.policy=compact")
    ```
  </Tab>

  <Tab value="Java">
    ```java
    import org.apache.kafka.clients.admin.Admin;
    import org.apache.kafka.clients.admin.NewTopic;

    import java.util.List;
    import java.util.Map;
    import java.util.Properties;

    public final class CreateCompactedTopic {
        public static void main(String[] args) throws Exception {
            Properties props = new Properties();
            props.put("bootstrap.servers", "localhost:9092");

            try (Admin admin = Admin.create(props)) {
                NewTopic topic = new NewTopic("user-profiles", 1, (short) 1)
                        .configs(Map.of("cleanup.policy", "compact"));
                admin.createTopics(List.of(topic)).all().get();
                System.out.println("created user-profiles with cleanup.policy=compact");
            }
        }
    }
    ```
  </Tab>

  <Tab value="JavaScript">
    ```javascript
    const { Kafka } = require("kafkajs");

    const kafka = new Kafka({ brokers: ["localhost:9092"] });
    const admin = kafka.admin();

    async function main() {
      await admin.connect();
      await admin.createTopics({
        topics: [
          {
            topic: "user-profiles",
            numPartitions: 1,
            replicationFactor: 1,
            configEntries: [{ name: "cleanup.policy", value: "compact" }],
          },
        ],
      });
      console.log("created user-profiles with cleanup.policy=compact");
      await admin.disconnect();
    }

    main().catch((err) => {
      console.error(err);
      process.exit(1);
    });
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    using Confluent.Kafka;
    using Confluent.Kafka.Admin;

    var config = new AdminClientConfig { BootstrapServers = "localhost:9092" };

    using var admin = new AdminClientBuilder(config).Build();
    await admin.CreateTopicsAsync(new[]
    {
        new TopicSpecification
        {
            Name = "user-profiles",
            NumPartitions = 1,
            ReplicationFactor = 1,
            Configs = new Dictionary<string, string> { ["cleanup.policy"] = "compact" },
        },
    });
    Console.WriteLine("created user-profiles with cleanup.policy=compact");
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    require "rdkafka"

    admin = Rdkafka::Config.new("bootstrap.servers" => "localhost:9092").admin

    admin.create_topic("user-profiles", 1, 1, { "cleanup.policy" => "compact" })
         .wait(max_wait_timeout_ms: 10_000)
    puts "created user-profiles with cleanup.policy=compact"

    admin.close
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    use rdkafka::admin::{AdminClient, AdminOptions, NewTopic, TopicReplication};
    use rdkafka::client::DefaultClientContext;
    use rdkafka::config::ClientConfig;

    #[tokio::main]
    async fn main() {
        let admin: AdminClient<DefaultClientContext> = ClientConfig::new()
            .set("bootstrap.servers", "localhost:9092")
            .create()
            .expect("admin client creation failed");

        let topic = NewTopic::new("user-profiles", 1, TopicReplication::Fixed(1))
            .set("cleanup.policy", "compact");

        admin
            .create_topics(&[topic], &AdminOptions::new())
            .await
            .expect("create_topics failed");
        println!("created user-profiles with cleanup.policy=compact");
    }
    ```
  </Tab>
</Tabs>

<Callout type="warn">
  **`compact` requires the `next` engine.** The Kafka connector itself only runs on `next` (see
  [Architecture](/connectors/kafka/concepts/architecture)), so in normal operation you'll never
  hit this path — but the admission gate is real: a `cleanup.policy` request the connector doesn't
  recognize, or a `compact` request on a store that somehow isn't on `next`, is rejected
  `INVALID_CONFIG`(40) at `CreateTopics` — no partial topic is created. Valid values are `delete`
  (default), `compact`, and `compact,delete`.
</Callout>

## Enable compaction on an existing topic [#enable-compaction-on-an-existing-topic]

To flip an existing `delete`-policy topic over to `compact` — for example, a topic you're
repurposing as a Kafka Streams changelog — use `IncrementalAlterConfigs`(44), the same admin API
[Capabilities](/connectors/kafka/reference/capabilities) lists as partial support (a subset of
configs is recognized; `cleanup.policy` is one of them):

```go
package main

import (
	"context"
	"log"

	"github.com/twmb/franz-go/pkg/kadm"
	"github.com/twmb/franz-go/pkg/kgo"
)

func main() {
	ctx := context.Background()

	cl, err := kgo.NewClient(kgo.SeedBrokers("localhost:9092"))
	if err != nil {
		log.Fatalf("client: %v", err)
	}
	defer cl.Close()

	admin := kadm.NewClient(cl)
	if _, err := admin.AlterTopicConfigs(ctx, []kadm.AlterConfig{
		{Op: kadm.SetConfig, Name: "cleanup.policy", Value: kadm.StringPtr("compact")},
	}, "user-profiles"); err != nil {
		log.Fatalf("alter configs: %v", err)
	}
	log.Println("user-profiles is now cleanup.policy=compact")
}
```

Every pinned client library maps to the same wire call, with two gaps worth knowing about:

| Client                   | Incremental alter-configs call                                                                                                                                                                                                                     |
| ------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| Go (franz-go/kadm)       | `admin.AlterTopicConfigs(ctx, configs, topic)`                                                                                                                                                                                                     |
| Python (confluent-kafka) | `admin.incremental_alter_configs([ConfigResource(...)])`                                                                                                                                                                                           |
| Java (kafka-clients)     | `admin.incrementalAlterConfigs(Map<ConfigResource, Collection<AlterConfigOp>>)`                                                                                                                                                                    |
| JavaScript (kafkajs)     | **No client method.** kafkajs's `admin.alterConfigs()` issues the older, whole-state `AlterConfigs`(33) request, which this connector doesn't advertise — use one of the other clients to alter an existing topic's `cleanup.policy` from Node.js. |
| C# (Confluent.Kafka)     | `adminClient.IncrementalAlterConfigsAsync(configs)`                                                                                                                                                                                                |
| Ruby (rdkafka)           | `admin.incremental_alter_configs(resources_with_configs)`                                                                                                                                                                                          |
| Rust (rdkafka)           | **No incremental variant.** `admin_client.alter_configs(...)` only issues the whole-state `AlterConfigs`(33) request — same gap as kafkajs.                                                                                                        |

## Produce keyed records and a tombstone [#produce-keyed-records-and-a-tombstone]

Compaction only makes sense on keyed records: the compactor groups by key and keeps the newest
value. A record with a **null value** and a non-null key is a **tombstone** — a delete marker for
that key. The example below produces two versions of `user-42` (so the older value is eligible for
compaction) and a tombstone for `user-7` (marking it for removal):

<Tabs groupId="language" items="['kcat', 'Go', 'Python', 'Java', 'JavaScript', 'C#', 'Ruby', 'Rust']">
  <Tab value="kcat">
    `-Z` tells `kcat` to treat an empty value (after the `-K` key separator) as `NULL` rather than an
    empty string — that's what makes the second line below a real tombstone, not a zero-length value:

    ```bash
    # key "user-42", two values -- the newer one supersedes the older under compaction
    printf 'user-42:v1\nuser-42:v2\n' | kcat -P -b localhost:9092 -t user-profiles -Z -K:

    # tombstone: key "user-7", NULL value
    echo "user-7:" | kcat -P -b localhost:9092 -t user-profiles -Z -K:
    ```
  </Tab>

  <Tab value="Go">
    ```go
    package main

    import (
    	"context"
    	"log"

    	"github.com/twmb/franz-go/pkg/kgo"
    )

    func main() {
    	ctx := context.Background()
    	cl, err := kgo.NewClient(kgo.SeedBrokers("localhost:9092"))
    	if err != nil {
    		log.Fatalf("client: %v", err)
    	}
    	defer cl.Close()

    	records := []*kgo.Record{
    		{Topic: "user-profiles", Key: []byte("user-42"), Value: []byte("v1")},
    		{Topic: "user-profiles", Key: []byte("user-42"), Value: []byte("v2")}, // same key, newer value
    		{Topic: "user-profiles", Key: []byte("user-7"), Value: nil},          // tombstone
    	}
    	if err := cl.ProduceSync(ctx, records...).FirstErr(); err != nil {
    		log.Fatalf("produce: %v", err)
    	}
    	log.Println("produced 2 versions of user-42 and a tombstone for user-7")
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python
    from confluent_kafka import Producer

    producer = Producer({"bootstrap.servers": "localhost:9092"})

    producer.produce("user-profiles", key="user-42", value="v1")
    producer.produce("user-profiles", key="user-42", value="v2")  # same key, newer value
    producer.produce("user-profiles", key="user-7", value=None)   # tombstone
    producer.flush(10)
    print("produced 2 versions of user-42 and a tombstone for user-7")
    ```
  </Tab>

  <Tab value="Java">
    ```java
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import org.apache.kafka.common.serialization.StringSerializer;

    import java.util.Properties;

    public final class ProduceCompacted {
        public static void main(String[] args) {
            Properties props = new Properties();
            props.put("bootstrap.servers", "localhost:9092");
            props.put("key.serializer", StringSerializer.class.getName());
            props.put("value.serializer", StringSerializer.class.getName());

            try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
                producer.send(new ProducerRecord<>("user-profiles", "user-42", "v1"));
                producer.send(new ProducerRecord<>("user-profiles", "user-42", "v2")); // same key, newer value
                producer.send(new ProducerRecord<>("user-profiles", "user-7", null));  // tombstone
                producer.flush();
                System.out.println("produced 2 versions of user-42 and a tombstone for user-7");
            }
        }
    }
    ```
  </Tab>

  <Tab value="JavaScript">
    ```javascript
    const { Kafka } = require("kafkajs");

    const kafka = new Kafka({ brokers: ["localhost:9092"] });
    const producer = kafka.producer();

    async function main() {
      await producer.connect();
      await producer.send({
        topic: "user-profiles",
        messages: [
          { key: "user-42", value: "v1" },
          { key: "user-42", value: "v2" }, // same key, newer value
          { key: "user-7", value: null },  // tombstone
        ],
      });
      console.log("produced 2 versions of user-42 and a tombstone for user-7");
      await producer.disconnect();
    }

    main().catch((err) => {
      console.error(err);
      process.exit(1);
    });
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    using Confluent.Kafka;

    var config = new ProducerConfig { BootstrapServers = "localhost:9092" };
    using var producer = new ProducerBuilder<string, string>(config).Build();

    await producer.ProduceAsync("user-profiles", new Message<string, string> { Key = "user-42", Value = "v1" });
    await producer.ProduceAsync("user-profiles", new Message<string, string> { Key = "user-42", Value = "v2" }); // same key, newer value
    await producer.ProduceAsync("user-profiles", new Message<string, string> { Key = "user-7", Value = null });  // tombstone

    producer.Flush(TimeSpan.FromSeconds(10));
    Console.WriteLine("produced 2 versions of user-42 and a tombstone for user-7");
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    require "rdkafka"

    producer = Rdkafka::Config.new("bootstrap.servers" => "localhost:9092").producer

    producer.produce(topic: "user-profiles", key: "user-42", payload: "v1").wait
    producer.produce(topic: "user-profiles", key: "user-42", payload: "v2").wait # same key, newer value
    producer.produce(topic: "user-profiles", key: "user-7", payload: nil).wait   # tombstone

    puts "produced 2 versions of user-42 and a tombstone for user-7"
    producer.close
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    use rdkafka::config::ClientConfig;
    use rdkafka::producer::{BaseProducer, BaseRecord, Producer};
    use std::time::Duration;

    fn main() {
        let producer: BaseProducer = ClientConfig::new()
            .set("bootstrap.servers", "localhost:9092")
            .create()
            .expect("producer creation failed");

        producer
            .send(BaseRecord::to("user-profiles").key("user-42").payload("v1"))
            .expect("send failed");
        producer
            .send(BaseRecord::to("user-profiles").key("user-42").payload("v2")) // same key, newer value
            .expect("send failed");
        producer
            .send(BaseRecord::to("user-profiles").key("user-7")) // tombstone: no .payload() call => NULL value
            .expect("send failed");

        producer.flush(Duration::from_secs(10)).expect("flush failed");
        println!("produced 2 versions of user-42 and a tombstone for user-7");
    }
    ```
  </Tab>
</Tabs>

Consume from the beginning right after producing, and you'll see **all three** records — the
original `v1`, the newer `v2`, and the tombstone:

```bash
kcat -C -b localhost:9092 -t user-profiles -o beginning -c 3
```

That's expected: compaction is an **asynchronous background job**, not something that happens at
produce time. The next section covers when it actually runs.

## What compaction does in the background [#what-compaction-does-in-the-background]

A background compactor periodically scans a compacted topic's log, keeps only the latest record per
key, and removes everything older for that key. On KubeMQ the compactor runs on a **fixed
background interval** (a leader-gated tick every few seconds), not gated on how "dirty" the log is.
Kafka's `min.cleanable.dirty.ratio` and `segment.ms` are still accepted and echoed back at
`DescribeConfigs` for tooling compatibility, but they do **not** change the scan cadence on KubeMQ
today. A tombstone isn't removed immediately either — it's kept for `delete.retention.ms` (default
`86400000`, 24 hours) so that a
consumer reading through the log has a window to observe the delete before the tombstone itself
disappears.

The one invariant that matters most for client code: &#x2A;*compaction never renumbers surviving
offsets.** For the example above, after a compaction pass runs:

| Offset | Key       | Value         | After compaction                                      |
| ------ | --------- | ------------- | ----------------------------------------------------- |
| 0      | `user-42` | `v1`          | Removed — superseded by offset 1                      |
| 1      | `user-42` | `v2`          | Kept — the latest value for `user-42`                 |
| 2      | `user-7`  | *(tombstone)* | Kept until `delete.retention.ms` elapses, then reaped |

A `Fetch` at offset 0 after compaction doesn't error and doesn't get handed a renumbered record — it
returns the next surviving offset (1), exactly like real Kafka's own compacted-topic behavior, which
every conformant Kafka client already knows how to handle.

## What compaction unlocks [#what-compaction-unlocks]

Compaction is a prerequisite, not a nice-to-have, for two large parts of the Kafka ecosystem:

* **Kafka Connect** — its internal config, offset, and status topics are compacted by convention;
  Connect refuses to start against a broker that can't honor `cleanup.policy=compact` on them.
* **Kafka Streams** — a stateful topology's changelog topics are compacted so that restoring state
  after a restart only has to replay the latest value per key, not the topic's entire history.

Because the KubeMQ connector recognizes and runs `cleanup.policy=compact` today, both tools work
against it without any special-casing on their side — they see a normal compacted Kafka topic.

## Related [#related]

<Cards>
  <Card title="Durability & Retention" href="/connectors/kafka/concepts/durability-and-retention" description="The acks contract, time/size retention, and the compaction model this page turns into a runnable procedure." />

  <Card title="Capabilities" href="/connectors/kafka/reference/capabilities" description="Compaction's exact support status alongside every other implemented Kafka API." />

  <Card title="Migrate from Kafka" href="/connectors/kafka/how-to/migrate-from-kafka" description="Bringing an existing Kafka workload's data and offsets onto KubeMQ — a separate, start-fresh playbook." />

  <Card title="Producing" href="/connectors/kafka/how-to/producing" description="Keys, headers, acks, and the idempotent producer — the mechanics behind every keyed record on this page." />
</Cards>