Apache Kafka

Apache Kafka ermöglicht es, Datenströme über eine verteilte Streaming-Plattform zu speichern und zu verarbeiten. Mit verschiedenen Schnittstellen können Daten in Kafka-Cluster geschrieben, Daten gelesen oder aus Drittsystemen im- oder exportiert werden. Durch die hohe Fehlertoleranz und Skalierbarkeit eignet sich Apache Kafka für große Datenmengen und Anwendungen z. B. im Big-Data-Umfeld.

Weitere Informationen zur Verwendung von Apache Kafka in einer lokalen Umgebung finden Sie hier: https://kafka.apache.org/quickstart.

Weitere Informationen zur Konfiguration und Integration von TLS-Zertifikaten in Apache Kafka Cluster finden Sie hier: http://kafka.apache.org/documentation/#security_ssl.

Um die beiden Adapter Apache Kafka Publisher bzw. Apache Kafka Subscriber zu verwenden, wird eine Konfigurationsdatei in der X4 Server-Installation benötigt.

Diese Konfigurationsdatei muss apachekafka-config.xml heißen und im Ordner X4DB/0 liegen. Beim Start des X4 Servers werden alle Informationen aus dieser Datei geladen.

Die folgende XML-Datei zeigt alle erforderlichen Elemente und Attribute in der richtigen Struktur:

Beispieldatei apachekafka-config.xml

CODE

<?xml version='1.0' encoding='UTF-8'?>
<ApacheKafkaConfiguration>
	<!-- Many clusters can be defined here -->
	<!-- general approach -->
	<Cluster id="ApacheKafka_broker_1" server="localhost:9092">
		<!-- Now the different clients that can work with this broker are defined: 
			For each ClientId, we define the following things: - The ClientId name: alias 
			to identify the client in X4 Processes. - groupId: the id of the group where 
			the consumer/publisher belongs to. -->
		<Client id="AK_Client_1" groupId="test"
			autoCommitIntervalMS="100" />
		<Client id="AK_Client_2" groupId="test"
			autoCommitIntervalMS="100" />
		<!-- Now, the different topics that clients can interact with for this 
			broker are defined -->
		<Topic filter="topic1" />
		<Topic filter="topic2" />
		<Topic filter="topic3" />
	</Cluster>
	<!-- cluster with ssl / tls configuration -->
	<Cluster id="ApacheKafka_broker_2" server="localhost:9093">
		<!-- Node that includes the SSL/TLS needed configuration for server -->
		<SslConfiguration privateKeyPassword="password">
			<!-- Server keystore password and server keystore path -->
			<KeyStore password="password"
				path="C:\\tmp\\ssl_new\\server.keystore.jks" />
			<!-- Server truststore password and server truststore path -->
			<TrustStore password="password"
				path="C:\\tmp\\ssl_new\\server.truststore.jks" />
		</SslConfiguration>
		<!-- Now the different clients that can work with this broker are defined: 
			For each ClientId, we define the following things: - The ClientId name: alias 
			to identify the client in X4 Processes. - groupId: the id of the group where 
			the consumer/publisher belongs to. -->
		<Client groupId="test" autoCommitIntervalMS="100"
			id="AK_Client_3" />
		<Client groupId="test" autoCommitIntervalMS="100"
			id="AK_Client_4" />
		<!-- Now, the different topics that clients can interact with for this 
			broker are defined -->
		<Topic filter="topic2" />
		<Topic filter="topic5" />
		<Topic filter="topic4" />
	</Cluster>
	<!-- cluster with optional client properties for subscriber and publisher -->
	<Cluster id="ApacheKafka_broker_3" server="localhost:9093">
		<Client id="TestClient" groupId="TestGroup"
			autoCommitIntervalMS="100">
			<!-- optional subscriber propertiers for this client -->
			<subscriberProperties>
				<Property key="some.subscriber.property.key" value="value" />
			</subscriberProperties>
			<!-- optional publisher propertiers for this client -->
			<publisherProperties>
				<Property key="some.publisher.property.key" value="value" />
			</publisherProperties>
		</Client> <!-- client now needs to be closed -->
		<!-- Now, the different topics that clients can interact with for this 
			broker are defined -->
		<Topic filter="topic2" />
		<Topic filter="topic5" />
		<Topic filter="topic4" />
	</Cluster>
</ApacheKafkaConfiguration>

Information zu subscriberProperties und publisherProperties

In der Konfigurationsdatei apachekafka-config.xml können optional für jeden Client folgende Eigenschaften gesetzt werden:

Die subscriberProperties können für den Apache Kafka Subscriber für die Operationen Assign und Subscribe gesetzt werden.
Die publisherProperties können für den Apache Kafka Publisher für die Operation Publish gesetzt werden.

Die Eigenschaften subscriberProperties und publisherProperties können jeweils für nur einen Client und für nur eine Operation gesetzt werden.

Hinweis:

Das Verhalten von subscriberProperties überschreibt den Standard-Wert, der im Element autoCommitIntervalMS gesetzt ist, wenn dieser erneut durch die subscriberProperties gesetzt wurde.

Beispieldatei für apachekafka-config.xml mit Eigenschaften von Subscriber und Publisher

CODE

<?xml version='1.0' encoding='UTF-8'?>
<ApacheKafkaConfiguration>
<Cluster id="ApacheKafka_broker_1" server="localhost:9092">
	<Client id="AK_Client_1" groupId="TestGroup" autoCommitIntervalMS="100">
	<!-- Propertiers of the Subscriber -->
			<subscriberProperties>
				<Property key="some.subscriber.property.key" value="value" />
			</subscriberProperties>
	</Client>
	<Topic filter="test"/>
</Cluster>
<Cluster id="ApacheKafka_broker_2" server="localhost:9093">
	<Client id="AK_Client_2" groupId="TestGroup" autoCommitIntervalMS="100">
	<!-- Propertiers of the Publisher -->
			<publisherProperties>
				<Property key="some.publisher.property.key" value="value" />
			</publisherProperties>
	</Client>
	<Topic filter="test"/>
</Cluster>
</ApacheKafkaConfiguration>

Im Folgenden sind die wichtigsten Elemente und Attribute näher erläutert:

Element / Attribut	Beschreibung
`Cluster`	Muss eindeutig sein. Definiert die erforderlichen Parameter, um die Verbindung zu einem bestimmten Kafka-Cluster zu konfigurieren. Im Attribut `id` befindet sich die Information, die für den Adapter-Parameter `clusterId` benötigt wird.
`server`	URL des Clusters. Unterstützt das TCP-Protokoll. Mögliche Werte. `hostname:port`
`SslConfiguration`	Definiert die SSL/TLS-Konfigurationsparameter.
`privateKeyPassword`	Passwort des Private Key in der Keystore-Datei.
`KeyStore`	Definiert, wo sich der zu verwendende Keystore befindet und welches Passwort hinterlegt ist.
`TrustStore`	Definiert, wo sich der zu verwendende Trust Store befindet und welches Passwort hinterlegt ist.
`ClientId`	Informationen zum Client, der verwendet wird, um Nachrichten zu einem bestimmten Topic zu veröffentlichen.
`groupId`	Definiert die Gruppen-ID, die verwendet wird, um die Verbindungen zu gruppieren.
`autoCommitIntervalMS`	Definiert die Häufigkeit der Commits in Millisekunden. Dieses Attribut wird für den Adapter Apache Kafka Subscriber benötigt.
`Topic`	Topic, in dem die Nachrichten veröffentlicht werden.
`filter`	Name des Topics
`subscriberProperties`	Eigenschaften, die für den Adapter Apache Kafka Subscriber gesetzt werden können. `ConsumerConfig values` werden in den Server Logs bei Start des Prozesses geloggt. Wenn Werte für die Eigenschaften gesetzt werden, die für die `subscriberProperties` nicht verfügar sind, wird ein Fehler ausgegeben. Weitere Informationen finden Sie in der offiziellen Apache Kafka Documentation (Subscriber und Consumer sind hier synonym). Beispiele für mögliche Werte bei Verwendung des Consumers: https://kafka.apache.org/22/documentation.html#consumerconfigs
`publisherProperties`	Eigenschaften, die für den Adapter Apache Kafka Publisher gesetzt werden können. `ProducerConfig values` werden in den Server Logs bei Start des Prozesses geloggt. Wenn Werte für die Eigenschaften gesetzt werden, die für die `publisherProperties` nicht verfügar sind, wird ein Fehler ausgegeben. Weitere Informationen finden Sie in der offiziellen Apache Kafka Documentation (Publisher und Producer sind hier synonym). Beispiele für mögliche Werte bei Verwendung des Producers: https://kafka.apache.org/22/documentation.html#producerconfigs

Die Werte für KeyStore, TrustStore und privateKeyPassword können auch über Adapterparameter gesetzt werden. Werte, die über Adapterparameter gesetzt werden, überschreiben die Werte aus der apachekafka-config.xml.