Topic Management

CreateTopicsRequest {
  topics: [
    {
      name: "orders"
      num_partitions: 3
      replication_factor: 1      // ignored in single-node setup
      assignments: []            // empty = auto-assign
      configs: [
        {name: "retention.ms", value: "86400000"},        // 1 day
        {name: "retention.bytes", value: "1073741824"},   // 1 GB
        {name: "compression.type", value: "lz4"},
        {name: "storage.compression", value: "zstd"}
      ]
    }
  ]
  timeout_ms: 60000
  validate_only: false         // true = validate but don't create
}

CreateTopicsResponse {
  throttle_time_ms: 0
  topics: [
    {
      name: "orders"
      error_code: 0              // NONE
      error_message: null
      num_partitions: 3
      replication_factor: 1
      configs: [
        {name: "retention.ms", value: "86400000", read_only: false, is_default: false},
        {name: "retention.bytes", value: "1073741824", read_only: false, is_default: false},
        {name: "compression.type", value: "lz4", read_only: false, is_default: false},
        {name: "storage.compression", value: "zstd", read_only: false, is_default: false}
      ]
    }
  ]
}

# Topic creation is primarily a broker-side operation
# Topic metadata (num_partitions, retention, compression, etc.) is stored
# in Redis hashes under the keyspace prefix

# Redis Streams for partitions are created lazily on first XADD:
# - korvet:orders:0 (created when first message produced to partition 0)
# - korvet:orders:1 (created when first message produced to partition 1)
# - korvet:orders:2 (created when first message produced to partition 2)

# No initialization commands needed - streams are auto-created by Redis

DeleteTopicsRequest {
  topic_names: ["orders", "payments"]
  timeout_ms: 30000
}

DeleteTopicsResponse {
  throttle_time_ms: 0
  topics: [
    {
      name: "orders"
      error_code: 0            // NONE
      error_message: null
    },
    {
      name: "payments"
      error_code: 3            // UNKNOWN_TOPIC_OR_PARTITION
      error_message: "Topic 'payments' does not exist"
    }
  ]
}

# 1. Verify topic exists in broker metadata (not Redis)
# If not exists: return UNKNOWN_TOPIC_OR_PARTITION (error code 3)

# 2. Get partition count from broker metadata
# For topic "orders" with 3 partitions

# 3. Destroy consumer groups for each partition
# For partition 0:
XGROUP DESTROY korvet:orders:0 my-group
XGROUP DESTROY korvet:orders:0 other-group
# For partition 1:
XGROUP DESTROY korvet:orders:1 my-group
XGROUP DESTROY korvet:orders:1 other-group
# For partition 2:
XGROUP DESTROY korvet:orders:2 my-group
XGROUP DESTROY korvet:orders:2 other-group
# Note: XGROUP DESTROY returns error if group doesn't exist - ignore errors

# 4. Delete partition streams
DEL korvet:orders:0
DEL korvet:orders:1
DEL korvet:orders:2

# 5. Remove topic from broker metadata (Redis hash)

# Alternative: Use SCAN to find all partition streams (safer)
SCAN 0 MATCH korvet:orders:* COUNT 100 TYPE stream
# Returns: cursor and list of stream keys
# Then DEL all found keys

MetadataRequest {
  topics: null                 // null = all topics, or specific topic list
  allow_auto_topic_creation: false
  include_cluster_authorized_operations: false
  include_topic_authorized_operations: false
}

MetadataResponse {
  throttle_time_ms: 0
  brokers: [
    {
      node_id: 0
      host: "localhost"
      port: 9092
      rack: null
    }
  ]
  cluster_id: "korvet-cluster-1"
  controller_id: 0
  topics: [
    {
      error_code: 0
      name: "orders"
      is_internal: false
      partitions: [
        {
          error_code: 0
          partition: 0
          leader: 0              // broker node_id
          replica_nodes: [0]
          isr_nodes: [0]         // in-sync replicas
          offline_replicas: []
        },
        {
          error_code: 0
          partition: 1
          leader: 0
          replica_nodes: [0]
          isr_nodes: [0]
          offline_replicas: []
        },
        {
          error_code: 0
          partition: 2
          leader: 0
          replica_nodes: [0]
          isr_nodes: [0]
          offline_replicas: []
        }
      ]
    },
    {
      error_code: 0
      name: "payments"
      is_internal: false
      partitions: [
        {
          error_code: 0
          partition: 0
          leader: 0
          replica_nodes: [0]
          isr_nodes: [0]
          offline_replicas: []
        }
      ]
    }
  ]
}

# Metadata requests are handled by broker's in-memory topic registry
# No Redis commands needed

# Broker returns topic metadata from memory/config:
# - Topic names
# - Partition counts
# - Retention policies
# - Compression settings
# - etc.

# Optional: Use SCAN to discover existing partition streams
SCAN 0 MATCH korvet:*:* COUNT 100 TYPE stream
# Returns: cursor and list of stream keys like ["korvet:orders:0", "korvet:orders:1", "korvet:payments:0"]
# Can be used to verify which topics/partitions actually have data