Wanted: topic management, declarative

[solsson]

We're a small team with a small set of kafka topics and clients, but already we need a self-documenting way to create and update topics.

Use cases:

• Ensure that a topic exists in minikube, or docker-compose, during development.
• Ensure that a topic exists in production during service provisionting in Kubernetes.
• Edit retention.ms.
• Knowing if the topic was created with key and/or value scemas in mind.

Using kafka-topics.sh or the java api ok for testing, but for instance with production topics we want to create them with 1 replica during dev and >=3 in a real ckluster.

An option is to use helm for topic creation manifests. I've searched for best practices, like tooling other organizations use, but with no luck. I imagine most teams will want to document their kafka topics, beyond a naming convention.

[BenjaminDavison]

We are thinking about using terraform for this: https://github.com/packetloop/terraform-provider-kafka

[DanielFerreiraJorge]

Maybe this: https://github.com/nbogojevic/kafka-operator

[solsson]

My current best bet is to rely on auto.create.topics.enable=true + default.replication.factor and a naming convention (some ideas here).

I found that during stream processing it's a quite legitimate use case to want to add new topics dynamically. For example in log processing you could split on kubernetes namespace, and those may pop up while your streams app is running.

[solsson]

I've tried to summarize my confusion by closing various open investigations. What surprised me the most was the insight(?) that Schema Registry doesn't help.

At Yolean we store all our persistent data in Kafka. Some of it has the integrity requirements you'd normally entrust a relational database. If validation is supported at the API level when producing, we'll make design choices based on the type of data (rather than time constraints / laziness).

Now I suspect we won't get that far without in-house libs - a maintenance effort we can't depend on, given the rapid evolution of Kafka APIs.

Validation can still be simple enough, if schemas are available as files. We already use json-schema in some core producers using Node.js. Files have the added benefit that they get the same versioning and release cycle as the services they share repository with.

We do need a little bit of tooling to copy schema files, be they Avro or JSON or Protobuf, to the source folders of the services that depend on them. Due to Docker build contexts we can't pull them from a parent folder.

We still don't enforce anything, but luckily there's a compelling alternative: Monitoring (explored in #49 and #93). Consumer services can export counters on the number of messages they had to drop due to invalidity, or even if they don't we can look at generic failure rates. We can catch errors that schemas can't: the other day we had misbehaving client-side code that led to an increase in messages/second to a topic by a factor 100.

We can also write dedicated validation services, whose sole purpose is to expose metrics. This is where naming conventions might matter. Let's say that topic names ending with a dash followed by digits indicate a version number. It'd be quite easy to spot, based on a topic listing and the bytes or message rates reported by broker JMX, that we have some service producing to an old version of a topic.

[solsson]

We're now phasing in auto.topic.create (#107) in production. The gotchas that made me disable it from the start are there for sure -- run kafkacat with some typo in -t and a new topic gets created. Who would configure their database to create a table on INSERT INTO xyz?

However, Kafka Streams in essence requires auto creation. We just have to live with the gotchas.

Regarding topics + schema management, with the arguments I tried to summarize above, I ended up writing a Gradle build that generates Java POJO typed Serde impls from json-schema files. We can also use these schema files directly from libs like ajv in Node.

To do this we had to establish a list of topics, or actually topic name regexes, where we configure the model (i.e. schema) that is used for values. With this list we can unit test the whole thing: "deserialize this byte stream for topic x" etc, which is what I argued for earlier.

Regarding topic management, with or without auto create, we could compare that list to existing topics in kafka, which together with bytes in/out rates could indicate if there are bogus topics lying around or we're writing to unexpected topics.