KubeMQ
ConnectorsKafkaHow-to guides

Compacted Topics

Run log-compacted topics on KubeMQ — set cleanup.policy=compact, understand tombstones and latest-value-per-key retention, and the ecosystems it unlocks.

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; this page is the practical companion.

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 rather than treating compaction as something you "migrate."

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.

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")
}
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")
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");
        }
    }
}
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);
});
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");
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
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");
}

compact requires the next engine. The Kafka connector itself only runs on next (see 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.

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 lists as partial support (a subset of configs is recognized; cleanup.policy is one of them):

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:

ClientIncremental 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

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):

-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:

# 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:
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")
}
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")
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");
        }
    }
}
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);
});
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");
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
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");
}

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

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

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: compaction never renumbers surviving offsets. For the example above, after a compaction pass runs:

OffsetKeyValueAfter compaction
0user-42v1Removed — superseded by offset 1
1user-42v2Kept — the latest value for user-42
2user-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

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.

Durability & Retention

The acks contract, time/size retention, and the compaction model this page turns into a runnable procedure.

Capabilities

Compaction's exact support status alongside every other implemented Kafka API.

Migrate from Kafka

Bringing an existing Kafka workload's data and offsets onto KubeMQ — a separate, start-fresh playbook.

Producing

Keys, headers, acks, and the idempotent producer — the mechanics behind every keyed record on this page.

Was this page helpful?

Consuming

Consume from KubeMQ over the Kafka protocol — join a consumer group, commit offsets manually or automatically, and seek by offset or timestamp.

Transactions & EOS

Exactly-once semantics on KubeMQ — the transactional producer, read_committed isolation, consume-transform-produce, and producer fencing.

On this page

Create a compacted topicEnable compaction on an existing topicProduce keyed records and a tombstoneWhat compaction does in the backgroundWhat compaction unlocksRelated