Logical topics

---
kind: ConcentrationRule
metadata:
  name: concentration1
spec:
  pattern: concentrated.*
  physicalTopics:
    delete: physical.topic-delete
    compact: physical.topic-compact
    # deleteCompact: physical.topic-deletecompact

kafka-topics --create 
    --bootstrap-server conduktor-gateway:6969 \
    --topic <your-topic-name> \
    --partitions 3 \
    --config cleanup.policy=compact,delete
Error while executing topic command : Cleanup Policy is invalid

kafka-topics --create 
    --bootstrap-server conduktor-gateway:6969 \
    --topic <your-topic-name> \
    --partitions 3 \
    --config retention.ms=704800000
Error while executing topic command : Value '704800000' for configuration 'retention.ms' is incompatible with physical topic value '604800000'.

---
kind: ConcentrationRule
metadata:
  name: concentration1
spec:
  pattern: concentrated.*
  physicalTopics:
    delete: physical.topic
  autoManaged: true

kafka-configs --bootstrap-server kafka:9092 \
    --entity-type topics --entity-name physical.topic \
    --describe
Configs for topic 'physical.topic' are:
  cleanup.policy=delete
  retention.ms=604800000
  retention.bytes=-1

kafka-topics --create 
    --bootstrap-server conduktor-gateway:6969 \
    --topic <your-topic-name> \
    --partitions 3 \
    --config retention.ms=704800000

kafka-configs --bootstrap-server kafka:9092 \
    --entity-type topics --entity-name physical.topic \
    --describe
Configs for topic 'physical.topic' are:
  cleanup.policy=delete
  retention.ms=704800000
  retention.bytes=-1

---
kind: ConcentrationRule
metadata:
  name: concentration1
spec:
  pattern: concentrated.*
  physicalTopics:
    delete: physical.topic
  offsetCorrectness: true

-- Rename fields
SELECT name AS employee_name, salary FROM employees

-- Select a subset of fields (others are dropped)
SELECT name, department FROM employees

-- Compute values with CASE expressions
SELECT name, CASE WHEN salary > 2000 THEN 'high' ELSE 'low' END AS salary_band FROM employees

-- Include record metadata alongside value fields
SELECT name, record.offset AS offset, record.partition AS partition FROM employees

-- Match email addresses
SELECT * FROM users WHERE email REGEXP '^[a-z]+@[a-z]+\.com$'

-- Match records where name starts with 'Bo'
SELECT * FROM customers WHERE name REGEXP 'Bo.*'

-- Match addresses containing 'Engineering'
SELECT * FROM employees WHERE education.major REGEXP '.*Engineering'

-- order_date holds an ISO-8601 date string, e.g. "2026-01-01"
SELECT * FROM orders WHERE order_date = {d '2026-01-01'}

-- event_at holds a timestamp string, space or T separator, fractional seconds optional:
-- "2026-01-01 12:34:56" or "2026-01-01T12:34:56.123"
SELECT * FROM events WHERE event_at > {ts '2026-01-01 12:34:56'}

-- alarm_at holds a time string, e.g. "12:00:00"
SELECT * FROM events WHERE alarm_at <= {t '12:00:00'}

-- Simple nested field
SELECT * FROM orders WHERE address.city = 'London'

-- Deeply nested field
SELECT * FROM users WHERE profile.settings.theme = 'dark'

-- Numeric comparison on nested field
SELECT * FROM metrics WHERE metadata.stats.count > 100

-- First element (index 0)
SELECT * FROM orders WHERE `items[0].name` = 'apple'

-- Last element (negative index)
SELECT * FROM orders WHERE `items[-1].name` = 'banana'

-- Numeric array access
SELECT * FROM metrics WHERE `values[1]` > 50

-- Nested array field
SELECT * FROM events WHERE `data.records[0].status` = 'active'

-- Value field, using the $ prefix for the payload
SELECT * FROM events WHERE $.severity = 'ERROR'

-- Raw record key, compared as a string
SELECT * FROM users WHERE record.key = 'account-42'

-- Nested key field when the key is JSON
SELECT * FROM users WHERE record.key.user.name = 'alice'

-- Header value, compared as a UTF-8 string
SELECT * FROM orders WHERE record.header.tenant = 'prod'

-- Partition, offset, or timestamp, compared as numbers
SELECT * FROM events WHERE record.timestamp > 1700000000000

-- These also work in the SELECT projection, not just WHERE filters
SELECT record.key AS key, record.header.tenant AS tenant, record.offset AS offset FROM orders

SELECT * FROM "vc-aliceorders" WHERE country = 'UK'

spec:
  transformation:
    type: sql
    statement: "SELECT * FROM customers WHERE country='UK'"
  onError:
    type: DROP

# Consume the uk_customers topic view
group.id=uk_customers::analytics

group.id=uk_customers::recent_orders::analytics

SELECT
    type,
    price as amount,
    color,
    CASE
        WHEN color = 'red' AND price > 1000 THEN 'Exceptional'
        WHEN price > 8000 THEN 'Luxury'
        ELSE 'Regular'
    END as quality,
    record.offset as record_offset,
    record.partition as record_partition
  FROM cars

{
  "name": "mySqlTopicPlugin",
  "pluginClass": "io.conduktor.gateway.interceptor.VirtualSqlTopicPlugin",
  "priority": 100,
  "config": {
    "virtualTopic": "legal_user",
    "statement": "SELECT * FROM users WHERE age > 18",
    "schemaRegistryConfig": {
       "host": "http://schema-registry:8081"
    }
  }
}

{
  "name": "myCelTopicPlugin",
  "pluginClass": "io.conduktor.gateway.interceptor.CelTopicPlugin",
  "priority": 100,
  "config": {
    "virtualTopic": "legal_user",
    "topic": "users",
    "expression": "record.value.age > 18",
    "schemaRegistryConfig": {
      "host": "http://schema-registry:8081"
    },
    "celCacheSize": 100
  }
}