Multityping in Kafka: Ein Praxisguide zur Nutzung eines Topics für verschiedene Event-Typen

In diesem Techup geht es darum, zu zeigen, wie man am besten mehrere Events in ein einzelnes Kafka-Topic schreibt. Dazu habe ich mir einen Anwendungsfall ausgedacht. Bei uns erscheinen regelmäßig Blogartikel zu verschiedenen Themen, die wir Techups nennen. Außerdem haben wir schon seit einiger Zeit unseren eigenen Podcast namens Decodify. All dies ist Teil unseres internen Forschungs- und Entwicklungsprozesses und wird in unserem Techhub gesammelt. Deshalb gibt es in diesem Beispiel ein Topic namens Techhub, das Events vom Typ Techup und Decodify empfängt.

Nachstehend ist zu sehen, dass unterschiedliche Events in das Topic TechHub geschrieben werden sollen:

Praktische Umsetzung: Ein Kafka-Topic für mehrere Event-Typen

Dabei wird das Beispiel mit einer Quarkus Applikation in der Version 3.9 umgesetzt. Dazu kann man am besten bei Quarkus selbst das Grundgerüst erstellen.

Schema-Entwicklung: Grundlagen und Implementierung

Im ersten Schritt müssen wir unsere gewünschten Schemas erstellen, aus denen anschliessend die entsprechenden Java Klassen erstellt werden sollen. Zunächst einmal das Techup-Schema — techup.avsc:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32


{
  "namespace": "com.bnova",
  "type": "record",
  "name": "Techup",
  "fields": [
    {
      "name": "title",
      "type": "string",
      "default": ""
    },
    {
      "name": "slug",
      "type": "string",
      "default": ""
    },
    {
      "name": "author",
      "type": "string",
      "default": ""
    },
    {
      "name": "content",
      "type": "string",
      "default": ""
    },
    {
      "name": "description",
      "type": "string",
      "default": ""
    }
  ]
}

Ebenfalls notwendig ist unser Decodify Schema — decodify.avsc:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40


{
  "namespace": "com.bnova",
  "type": "record",
  "name": "Decodify",
  "fields": [
    {
      "name": "episode",
      "type": "int",
      "doc": "The episode number.",
      "default": 0
    },
    {
      "name": "season",
      "type": "int",
      "doc": "The season number.",
      "default": 0
    },
    {
      "name": "topic",
      "type": "string",
      "doc": "The main topic of the episode.",
      "default": ""
    },
    {
      "name": "attendees",
      "type": {
        "type": "array",
        "items": "string"
      },
      "doc": "A list of attendees or participants in the episode.",
      "default": []
    },
    {
      "name": "description",
      "type": "string",
      "doc": "A brief description of the episode.",
      "default": ""
    }
  ]
}

Damit es jedoch möglich ist, mehrere Einträge in das selbe Topic zu schreiben, benötigen wir noch ein weitere Datei, welche als eine Referenz auf die anderen beiden Schemas genutzt wird.

bnova_techup_techhub_topic_all_types.avsc:

1
2
3
4


[
  "com.bnova.Techup",
  "com.bnova.Decodify"
]

Kafka Topics — Set-Up

Nun wird es Zeit das entsprechende Topic zu erstellen. Unser Techhub Topic muss innerhalb unseres Beispiel Projekts als Input und Output Topic definiert werden. Dies hat den Grund, dass wir später zu Testszwecken einen Endpunkt bereitstellen der in dieses Topic schreiben soll, damit wir testen können, dass auch beide Schemas ausgelesen werden können. Mittels dem Prefix mp.messaging.outgoing bzw. mp.messaging.incoming wird definiert, ob das Topic zum empfangen oder senden genutzt wird.

In unserem Beispiel wird der SmallRye Kafka Connector verwendet wird, um Nachrichten an Kafka zu senden und empfangen. Dies ermöglicht es der Anwendung, Nachrichten effizient an ein Kafka-Topic zu senden.

Die Einstellung mit dem Postfix topic definiert dabei den Namen des entsprechenden Topics. Zu dem muss der gewünschte Serializer und Deserializer definiert werden, damit die Daten korrekt verarbeitet werden können.

Damit es dem Deserializer möglich ist, das genaue Schema zu verwenden, das beim Schreiben der Daten verwendet wurde, wird specific.avro.reader aktiviert.

1
2
3
4
5
6
7
8
9


mp.messaging.outgoing.techhub-topic.connector=smallrye-kafka
mp.messaging.outgoing.techhub-topic.topic=${bnova.techup.topic.prefix}.techhub-topic.v1
mp.messaging.outgoing.techhub-topic.value.serializer=io.confluent.kafka.serializers.KafkaAvroSerializer

mp.messaging.incoming.techhub.connector=smallrye-kafka
mp.messaging.incoming.techhub.topic=${bnova.techup.topic.prefix}.techhub-topic.v1
mp.messaging.incoming.techhub.group.id=${bnova.techup.topic.prefix}-producer
mp.messaging.incoming.techhub.value.deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer
mp.messaging.incoming.techhub.specific.avro.reader=true

Essentielle Kafka-Services

Um verschieden Schemas zu nutzen benötigen wir ein Schema Registry. Dazu habe ich mich für die Lösung von Confluent entschieden. Zusammen bilden diese Services eine vollständige Kafka-Architektur, die über Docker bereitgestellt wird, und bieten eine robuste Plattform für die Nachrichtenübertragung und Schema-Verwaltung. Jeder Service spielt dabei eine spezifische Rolle:

Der Kafka Broker ist das Herzstück des Systems und verantwortlich für die Verwaltung und Speicherung von Nachrichten. Die Konfiguration legt die notwendigen Ports, Umgebungsvariablen und Listener fest, um die Kommunikation zu ermöglichen.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26


  broker:
    image: confluentinc/cp-kafka:7.6.0
    hostname: broker
    container_name: broker
    ports:
      - "9092:9092"
      - "9101:9101"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092'
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_JMX_PORT: 9101
      KAFKA_JMX_HOSTNAME: localhost
      KAFKA_PROCESS_ROLES: 'broker,controller'
      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@broker:29093'
      KAFKA_LISTENERS: 'PLAINTEXT://broker:29092,CONTROLLER://broker:29093,PLAINTEXT_HOST://0.0.0.0:9092'
      KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
      KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'
      # Replace CLUSTER_ID with a unique base64 UUID using "bin/kafka-storage.sh random-uuid"
      # See https://docs.confluent.io/kafka/operations-tools/kafka-tools.html#kafka-storage-sh
      CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qk'

Die Schema Registry von Confluent, spezifiziert im zweiten Service-Block, ist entscheidend für das Schema-Management, das in Kafka-Umgebungen benötigt wird, um die Kompatibilität und Versionierung der Schemas zu gewährleisten, die für die Nachrichtenserialisierung verwendet werden. Die Konfiguration gewährleistet, dass die Registry über die nötigen Informationen verfügt, um mit dem Kafka Broker zu kommunizieren und Schemas effektiv zu speichern und zu verwalten.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12


  schema-registry:
    image: confluentinc/cp-schema-registry:7.6.0
    hostname: schema-registry
    container_name: schema-registry
    depends_on:
      - broker
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'broker:29092'
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081

Um die Benutzerfreundlichkeit zu erhöhen und eine grafische Oberfläche für die Interaktion mit dem Kafka-Cluster zu bieten, wird der Service kafka-ui eingesetzt. Dieses Tool ermöglicht es den Benutzern, die Themen, Partitionen, Nachrichten und Schemas innerhalb des Clusters visuell zu inspizieren und zu verwalten. Es ist besonders nützlich für das Debugging und Monitoring, da es einen schnellen Überblick und einfache Steuerung der Kafka-Ressourcen ohne komplizierte Befehlszeilenoperationen bietet.

1
2
3
4
5
6
7
8
9


  kafka-ui:
    container_name: kafka-ui
    image: provectuslabs/kafka-ui:latest
    ports:
      - 8082:8080
    environment:
      DYNAMIC_CONFIG_ENABLED: 'true'
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: broker:29092
      KAFKA_CLUSTERS_0_NAME: 'local'

Registrierung und Nutzung von Avro-Schemas

Nachdem die Schema Registry eingerichtet ist, müssen die vorbereiteten Schemas registriert werden. Dies geschieht durch die Konfiguration in den application.properties, wo die URL der Registry wie folgt angegeben wird:

1

kafka.schema.registry.url=http://localhost:8081

Um die Schemas effektiv zu registrieren und verwalten, ist ein zusätzliches Plugin erforderlich. Die Avro-Schemas sollten in einer spezifischen Struktur im Projektverzeichnis abgelegt werden, damit die Abhängigkeiten / Referenzen aufgelöst werden können.

1
2
3
4
5
6
7
8


├── src
│   ├── main
│   │   ├── avro
│   │   │   ├── bnova_techup_techhub_topic_all_types.avsc
│   │   │   └── include
│   │   │       ├── decodify.avsc
│   │   │       └── techup.avsc
│   │   │

Das kafka-schema-registry-maven-plugin von Confluent wird in der pom.xml-Datei des Projekts konfiguriert. Dieses Plugin ermöglicht es, die Schemas direkt aus der Projektstruktur zu registrieren und sicherzustellen, dass sie mit den im Schema Registry definierten Kompatibilitätsrichtlinien übereinstimmen. Die Konfiguration des Plugins sieht wie folgt aus:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42


<plugin>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-schema-registry-maven-plugin</artifactId>
    <version>7.6.0</version>
    <configuration>
        <schemaRegistryUrls>
            <param>${kafka-schema-registry.url}</param>
        </schemaRegistryUrls>
        <subjects>
            <decodify>src/main/avro/include/decodify.avsc</decodify>
            <techup>src/main/avro/include/techup.avsc</techup>
            <bnova-techup.techhub-topic.v1-value>src/main/avro/bnova_techup_techhub_topic_all_types.avsc</bnova-techup.techhub-topic.v1-value>
        </subjects>
        <outputDirectory>src/main/avro/include</outputDirectory>
        <schemaTypes>
            <decodify>AVRO</decodify>
            <techup>AVRO</techup>
            <bnova-techup.techhub-topic.v1-value>AVRO</bnova-techup.techhub-topic.v1-value>
        </schemaTypes>
        <references>
            <bnova-techup.techhub-topic.v1-value>
                <reference>
                    <name>com.bnova.Decodify</name>
                    <subject>decodify</subject>
                </reference>
                <reference>
                    <name>com.bnova.Techup</name>
                    <subject>techup</subject>
                </reference>
            </bnova-techup.techhub-topic.v1-value>
        </references>
        <compatibilityLevels/>
        <messagePath/>
        <outputPath/>
        <previousSchemaPaths/>
        <schemas/>
    </configuration>
    <goals>
        <goal>register</goal>
        <goal>validate</goal>
    </goals>
</plugin>

Um sicherzustellen, dass die erforderlichen Dependencies verfügbar sind, müssen die Confluent-Repositories zur pom.xml hinzugefügt werden. Dies ermöglicht Maven, die spezifischen Confluent-Pakete zu finden und zu laden:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16


	<repositories>
		<!-- io.confluent:kafka-json-schema-serializer is only available from this repository: -->
		<repository>
			<id>confluent</id>
			<url>https://packages.confluent.io/maven/</url>
			<snapshots>
				<enabled>false</enabled>
			</snapshots>
		</repository>
	</repositories>
	<pluginRepositories>
		<pluginRepository>
			<id>confluent</id>
			<url>https://packages.confluent.io/maven/</url>
		</pluginRepository>
	</pluginRepositories>

Mit dieser Konfiguration ist man in der Lage, die Avro-Schemas effizient zu registrieren und sicherzustellen, dass sie korrekt validiert und kompatibel mit den Anforderungen der Kafka-Anwendungen sind.

Implementierung des Consumers

Nun fokussieren wir uns auf die Implementierung eines Kafka Consumers. Der Consumer wird durch eine Java-Klasse repräsentiert, die wir schlichtweg Consumer nennen. Um auf ein spezifisches Kafka-Topic zu hören, verwenden wir die Annotation @Incoming von MicroProfile Reactive Messaging. Hiermit wird angegeben, dass diese Methode Nachrichten aus dem Topic „techhub“ empfangen soll.

Für die Flexibilität im Umgang mit verschiedenen Schemata verwenden wir den SpecificRecord als Parameter der process-Methode. SpecificRecord ist eine Schnittstelle aus dem Apache Avro Framework, die es ermöglicht, generierte Avro-Objekte zu handhaben, die jeweils ein spezifisches Schema repräsentieren.

Aktuell ist die Funktionalität des Consumers darauf beschränkt, den eine Infos des empfangenen Avro-Records zu loggen. Die Klasse ist so konfiguriert, dass sie den Namen des Java-Klassenobjekts, das den Avro-Record repräsentiert, im Log ausgibt. In unserem Szenario erwarten wir, dass die Namen com.bnova.Techup und com.bnova.Decodify im Log erscheinen, je nachdem, welches Schema in den eingehenden Nachrichten verwendet wird.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35


package com.bnova.consumer;

import lombok.extern.jbosslog.JBossLog;

import org.apache.avro.specific.SpecificRecord;
import org.eclipse.microprofile.reactive.messaging.Incoming;
import com.bnova.Decodify;
import com.bnova.Techup;

@JBossLog
public class Consumer
{
	@Incoming("techhub")
	public void process(SpecificRecord record)
	{
		log.info(record.getClass().getName());

		if (record instanceof Techup techup)
		{
			log.info(techup.getTitle());
			log.info(techup.getSlug());
			log.info(techup.getAuthor());
			log.info(techup.getContent());
			log.info(techup.getDescription());
		}
		else if (record instanceof Decodify decodify)
		{
			log.info(decodify.getEpisode());
			log.info(decodify.getSeason());
			log.info(decodify.getTopic());
			log.info(decodify.getAttendees());
			log.info(decodify.getDescription());
		}
	}
}

Interaktives Testen der Topics

Um das Schreiben von Nachrichten in das Kafka-Topic „Techhub“ für Testzwecke zu ermöglichen, habe ich einen REST-Endpunkt implementiert, der es erlaubt, manuell Einträge für zwei unterschiedliche Objekttypen – Techup und Decodify – zu erstellen und zu senden. Der Endpunkt bietet zwei spezifische Pfade: /test/techup/{title} für ein Techup-Objekt und /test/decodify/{title} für ein Decodify-Objekt.

Für das Senden dieser Objekte ins Kafka-Topic wird der Emitter-Mechanismus von MicroProfile Reactive Messaging verwendet. Dieser Emitter wird mittels der Annotation @Channel mit dem Namen des Topics verbunden, in das geschrieben werden soll. Hierdurch ist es möglich, Nachrichten direkt aus der Anwendung heraus an das spezifizierte Topic zu senden.

Im Java-Code sieht das wie folgt aus:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53


package com.bnova.endpoint;

import jakarta.inject.Inject;
import jakarta.ws.rs.GET;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.PathParam;

import java.util.List;
import org.apache.avro.specific.SpecificRecord;
import org.eclipse.microprofile.reactive.messaging.Channel;
import org.eclipse.microprofile.reactive.messaging.Emitter;
import com.bnova.Decodify;
import com.bnova.Techup;

@Path(value = "test")
public class ExampleGenerator
{
	@Inject
	@Channel("techhub-topic")
	Emitter<SpecificRecord> commandTopicEmitter;

	@Path("techup/{title}")
	@GET
	public void createExampleTechup(@PathParam("title") String title)
	{
		var techup = Techup
				.newBuilder()
				.setTitle(title)
				.setSlug(title + "-slug")
				.setAuthor("Ricky")
				.setContent("example content for " + title)
				.setDescription("example desciption for " + title)
				.build();

		commandTopicEmitter.send(techup);
	}

	@Path("decodify/{episode}")
	@GET
	public void createExamplePodcast(@PathParam("episode") int eNr)
	{
		var episode = Decodify
				.newBuilder()
				.setEpisode(eNr)
				.setSeason(1)
				.setTopic("Multiple Eventtypes in the same topic")
				.setAttendees(List.of("Ricky", "Stefan", "Tom", "Wasili"))
				.setDescription("This is a decodify episode to talk about multiple diffrent event types in the same Kafka topic")
				.build();

		commandTopicEmitter.send(episode);
	}
}

Durch die Implementierung dieser Endpunkte kann der Nutzer durch einfache HTTP GET-Anfragen beispielhafte Techup und Decodify Objekte erstellen und an das Kafka-Topic senden. Diese Methode eignet sich hervorragend für das Debugging und Testing von Integrations- und Verarbeitungslogiken in der Kafka-Umgebung.

Inbetriebnahme

Um die benötigte Infrastruktur für unsere Kafka- und Quarkus-Anwendung zu testen, starten wir zuerst die Services über Docker Compose. Durch den Befehl docker compose up im Verzeichnis mit unserer docker-compose.yaml-Datei werden alle notwendigen Images heruntergeladen und die Container gestartet. Die korrekte Ausführung und den Status der Container können wir anschließend mit dem Befehl docker ps überprüfen. Das Ergebnis sollte in etwa so aussehen:

1
2
3
4


CONTAINER ID   IMAGE                                   COMMAND                  CREATED              STATUS              PORTS                                            NAMES
ddf93697afd8   confluentinc/cp-schema-registry:7.6.0   "/etc/confluent/dock…"   About a minute ago   Up About a minute   0.0.0.0:8081->8081/tcp                           schema-registry
e28aa00fb7fe   provectuslabs/kafka-ui:latest           "/bin/sh -c 'java --…"   About a minute ago   Up About a minute   0.0.0.0:8082->8080/tcp                           kafka-ui
3bce38d889ed   confluentinc/cp-kafka:7.6.0             "/etc/confluent/dock…"   About a minute ago   Up About a minute   0.0.0.0:9092->9092/tcp, 0.0.0.0:9101->9101/tcp   broker

Als Nächstes überprüfen wir, ob die Schema Registry erfolgreich gestartet wurde und zugänglich ist, wobei zu diesem Zeitpunkt noch keine Schemas registriert sein sollten.

Sobald die Services laufen, starten wir unsere Quarkus-Anwendung im Entwicklungsmodus mit mvn quarkus:dev. Ist dieser Vorgang erfolgreich, sollten wir in der Kafka-UI das neu angelegte Topic bnova-techup.techhub-topic.v1 sehen können.

Nun ist es Zeit, die Schemas für unsere Events zu registrieren, die wir zuvor in unserem Maven-Projekt konfiguriert haben. Dies erreichen wir durch den Ausführungsbefehl mvn schema-registry:register, der die Schemas im Schema Registry registriert:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17


~/Development/techhub/quarkus-kafka/producer ❯ mvn schema-registry:register                                                                                                                                                                                                                          16:02:59
[INFO] Scanning for projects...
[INFO] 
[INFO] -------------------------< com.bnova:producer >-------------------------
[INFO] Building producer 1.0.0-SNAPSHOT
[INFO] --------------------------------[ jar ]---------------------------------
[INFO] 
[INFO] --- kafka-schema-registry-maven-plugin:7.6.0:register (default-cli) @ producer ---
[INFO] Registered subject(decodify) with id 1 version 1
[INFO] Registered subject(techup) with id 2 version 1
[INFO] Registered subject(bnova-techup.techhub-topic.v1-value) with id 3 version 1
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time:  1.564 s
[INFO] Finished at: 2024-04-10T16:03:03+02:00
[INFO] ------------------------------------------------------------------------

Die erfolgreiche Registrierung kann auch direkt in der Registry überprüft werden.

Der eigentliche Test erfolgt durch das Senden von zwei unterschiedlichen Event-Typen über unseren REST-Endpunkt. Wir verwenden hierfür Curl-Befehle, um jeweils ein Decodify- und ein Techup-Event ins Topic zu senden.

Wie man auf der rechten Seite sehen kann, wurde jeweils ein decofiy-Event und ein Techup-Event gesendet. Auf der linken sieht man den Output, der wie gewünscht die korrekten Klassen ausgibt.

Die Ergebnisse dieser Aktionen können wir nicht nur in der Konsole sehen, sondern auch in der Kafka-UI, wo die eingegangenen Nachrichten dargestellt sind.

[!TIP]

Der gesamte Code kann auch über user Techhub Repository angeschaut werden

Fazit

In der Umsetzung dieses Ansatzes treten sowohl Herausforderungen als auch deutliche Vorteile auf. Technisch gesehen erhöht die Vereinfachung durch ein einzelnes Topic zunächst die Komplexität, insbesondere während der anfänglichen Konfiguration. Ein spezifischer Typ wird nicht direkt aus dem Topic ausgelesen, sondern als SpecificRecord verarbeitet, der vor der Weiterverarbeitung überprüft werden muss. Die Nutzung von Avro-Schemas schränkt zudem die Verwendung generischer Klassen oder Vererbungsstrukturen ein.

Auf der positiven Seite steht die signifikante Reduktion von Verwaltungsaufwand und Kosten, da weniger Topics gepflegt werden müssen und die Kosten oft pro Topic berechnet werden. Zudem erlaubt die Struktur eine flexible Erweiterung um neue Typen ohne großen Aufwand: Es sind lediglich das Erstellen eines neuen Schemas, dessen Registrierung und eventuell die Anpassung der Businesslogik notwendig. Dieser Ansatz bietet also eine effiziente Lösung für dynamische und skalierbare Datenarchitekturen in modernen Anwendungen.