Log Compaction

# Create a compacted topic
kafka-topics.sh --bootstrap-server localhost:9092 \
    --create \
    --topic user-profiles \
    --partitions 6 \
    --replication-factor 3 \
    --config cleanup.policy=compact \
    --config min.insync.replicas=2 \
    --config min.compaction.lag.ms=60000  # records survive at least 60s before compaction

# Compaction behaviour example for key "user-123":
# Offset 0: key=user-123, value={"name":"Alice","email":"a@ex.com"}
# Offset 5: key=user-123, value={"name":"Alice","email":"alice@ex.com"}  ← UPDATE
# Offset 9: key=user-123, value={"name":"Alice Smith","email":"alice@ex.com"} ← UPDATE
#
# After compaction — log retains only:
# key=user-123, value={"name":"Alice Smith","email":"alice@ex.com"} (offset 9)
# Offsets 0 and 5 are discarded.

# Convert an existing topic to compacted:
kafka-configs.sh --bootstrap-server localhost:9092 \
    --entity-type topics --entity-name user-profiles \
    --alter --add-config cleanup.policy=compact

// Producer — publish tombstone to delete a key
KafkaProducer<String, String> producer = new KafkaProducer<>(props);

// Tombstone: null value signals "delete this key"
ProducerRecord<String, String> tombstone =
    new ProducerRecord<>("user-profiles", "user-123", null);
producer.send(tombstone);

// After delete.retention.ms, the key "user-123" no longer exists in the topic
// Consumers who process the tombstone know to delete user-123 from their local store

// Spring Kafka — sending tombstone
@Service
public class UserProfileService {
    private final KafkaTemplate<String, UserProfile> kafka;

    public void deleteUser(String userId) {
        // KafkaTemplate with null value = tombstone
        kafka.send("user-profiles", userId, null);
    }

    public void updateUser(UserProfile profile) {
        kafka.send("user-profiles", profile.getId(), profile);
    }
}

// Kafka Streams KTable — backed by compacted changelog topic
StreamsBuilder builder = new StreamsBuilder();

// A KTable materialized view: latest user profile per user_id
KTable<String, UserProfile> userProfiles =
    builder.table(
        "user-profiles",   // compacted topic
        Materialized.as("user-profiles-store")
    );

// Join a stream with the KTable (enrich order events with user data)
KStream<String, Order> orders = builder.stream("orders");
KStream<String, EnrichedOrder> enriched = orders.join(
    userProfiles,
    (order, profile) -> new EnrichedOrder(order, profile.getEmail())
);
enriched.to("enriched-orders");

// On startup, Kafka Streams rehydrates the KTable store by replaying
// the compacted user-profiles topic from offset 0 — gets all current profiles
// without needing to replay years of full history.