Configure SQL

​

Overview

• Querying Kafka message data
• Querying Kafka metadata (such as the offset, partition and timestamp)

​

Console configuration

​

Database configuration

• CDK_KAFKASQL_DATABASE_URL: database connection URL in the format [jdbc:]postgresql://[user[:password]@]netloc[:port][/dbname][?param1=value1&...]. If you don’t have this set, SQL will not appear in the sidebar.

• CDK_KAFKASQL_DATABASE_HOST: Postgresql server host name
• CDK_KAFKASQL_DATABASE_PORT: Postgresql server port
• CDK_KAFKASQL_DATABASE_NAME: Database name
• CDK_KAFKASQL_DATABASE_USERNAME: Database login role
• CDK_KAFKASQL_DATABASE_PASSWORD: Database login password
• CDK_KAFKASQL_DATABASE_CONNECTIONTIMEOUT: Connection timeout option in seconds

​

Additional configuration

• CDK_KAFKASQL_CONSUMER-GROUP-ID: Consumer group name for the indexing process (default is conduktor-sql)
• CDK_KAFKASQL_CLEAN-EXPIRED-RECORD-EVERY-IN-HOUR: The interval in which the clean-up process will run to purge data outside the desired retention period
• CDK_KAFKA_SQL_REFRESH-TOPIC-CONFIGURATION-EVERY-IN-SEC: Time interval before taking into account any sqlStorage configuration changes on Topic resources
• CDK_KAFKASQL_REFRESH-USER-PERMISSIONS-EVERY-IN-SEC: See RBAC

​

Index topics for querying

​

Index topics in the UI

• Indexed Retention Time: The furthest point in time from which the data will be indexed. Note that any data before this point in time will be purged periodically.
  • By default, purging happens every 1h, but it’s configurable using the environment variable CDK_KAFKASQL_CLEAN-EXPIRED-RECORD-EVERY-IN-HOUR

• Offset lag: The difference between the latest message offset in the Kafka topic and the current offset of the consumer group, indicating how much data is yet to be processed
• Time lag: The delay between the timestamp of the latest message in the Kafka topic and the time when the message was indexed, reflecting processing latency
• Indexed count: The total number of messages successfully indexed into the database table from the Kafka topic

​

Index Topics in the CLI

---
apiVersion: kafka/v2
kind: Topic
metadata:
  cluster: kafka-cluster-dev
  name: customers
  sqlStorage:
    retentionTimeInSecond: 86400 # 1 day of retention
    enabled: true
spec:
  replicationFactor: 1
  partitions: 1
  configs:
    cleanup.policy: delete

conduktor apply -f topics.yml

​

Querying the data

SELECT *
  FROM "kafka-cluster-dev_customers";

conduktor sql 'select * from "kafka-cluster-dev_customers"' -n 2

​

Database storage format

• __timestamp
• __partition
• __offset
• __checksum: the checksum of the tuple (message(s) key, message(s) value)
• __headers: the message(s) headers
• __schema_id: the schema id
• __key: the message(s) key

{
    "a": {
        "b": {
            "c": "Hello World"
        },
        "userId": "109210921092"
    }
}

{
    "newField": "Kafka"
}

{
    "my_array": [
        {
            "something": "foo"
        },
        {
            "something_else": "bar"
        }
    ]
}

​

Shrinker

​

Collision Solver

• _table_mappings
• _table_fields_mappings

​

Database isolation

• Read-Only Connections: While not foolproof, enforcing read-only connections limits the potential for data modification
• SQL query pre-parsing and sanitizing:
  • Schema restriction: Restricting queries to the public schema prevents access to sensitive data in other schemas. For example, in the Conduktor database, the public schema is empty (except for the Flyway migration table which is also hidden)
  • Query Type Limitation: Allowing only SELECT statements ensures that users cannot modify or delete data. For example, it forbids ROLLBACK which would break the previous limitation

• Resource Contention: Prevents the Kafka indexing process or a user’s arbitrary request from consuming excessive resources and impacting the overall system performance
• Data Breach Mitigation: Limits the potential damage in case of a security breach in the SQL endpoint protection (not totally foolproof).

​

SQL security

​

RBAC

1. Initial Role Creation: When a user executes a SQL query on Console for the first time, the system checks if the corresponding user ROLE exists in the PostgreSQL database. If it does not, the RBAC system is consulted to determine the topics for which the user has the kafka.topics.read permission. The user’s ROLE is then created in PostgreSQL, and read access to the relevant topic tables is granted.
2. Periodic Role Updates: A background process in the Console periodically updates the access rights for each user ROLE to ensure they remain aligned with the RBAC permissions. The job is run every 30 seconds by default and can be customized by setting the env variable CDK_KAFKASQL_REFRESHUSERPERMISSIONSEVERYINSEC.

​

Data masking

• Instead of applying the masking policy as defined in Console, access to the entire column is restricted for simplicity
• For JSON blob columns, we are unable to restrict access to sub-objects, so access to the entire column content is restricted instead
• If access to a column is denied, the user cannot use a wildcard in a SELECT query (e.g., SELECT * FROM table). Attempting to do so will result in an access denied error

​

UI Experience

​

Known limitations

• Data formats currently supported are plain JSON, and both Avro & JSON with Confluent Schema Registry
• If for any reason a record can’t be parsed, they are ignored and the consumer continues
• To efficiently import data in Postgres, we didn’t set any primary key, so a record can be there more than once
• If you try to index a topic with a schema that is not supported, the lag value will be 0 but no records will appear in the table
• Admin API tokens are not supported with the Conduktor CLI. To execute SQL using the CLI, please utilize the user password authentication method, or short-lived user token.