Utilisation des API Kafka

Configurez Apache Kafka pour la compatibilité avec les API avec Oracle Cloud Infrastructure Streaming. Lorsque vos producteurs utilisent les API Kafka pour interagir avec le service de flux en continu, la décision de savoir vers quelle partition publier un message unique est gérée côté client par Kafka.

Les informations suivantes fournissent des exemples Java. Pour des exemples Python, voir Client Python Kafka et démarrage rapide du service de diffusion en continu.

Pour plus d'informations, voir Prise en charge des API Kafka.

Points d'extrémité

Pour les serveurs d'amorçage, utilisez le point d'extrémité de votre région sur le port 9092. Par exemple :

streaming.us-phoenix-1.oci.oraclecloud.com:9092

Authentification

L'authentification avec le protocole Kafka utilise des jetons d'authentification et le mécanisme SASL/PLAIN. Vous pouvez générer des jetons dans la page des détails de l'utilisateur de la console. Pour plus d'informations, voir Utilisation de jetons d'authentification.

Conseil

Créez un groupe/utilisateur dédié et accordez à ce groupe l'autorisation de gérer les flux dans le compartiment ou la location approprié. La politique sous Permettre aux administrateurs de gérer les ressources de diffusion en continu permet au groupe indiqué de tout faire avec la diffusion en continu et les ressources connexes du service de diffusion en continu. Vous pouvez ensuite générer un jeton d'authentification pour l'utilisateur que vous avez créé et l'utiliser dans votre configuration de client Kafka.

Votre nom d'utilisateur doit avoir le format suivant :

tenancyName/domain/username/streamPoolId

Conseil

Si vous utilisez la trousse SDK pour Java, vous pouvez également utiliser une autorisation du principal d'instance.

Configuration Kafka

Définissez les propriétés suivantes pour votre client Kafka.

Pour la trousse SDK Java

Paramètres recommandés pour la trousse SDK Java :

Properties properties = new Properties();
	properties.put("bootstrap.servers", "streaming.{region}.oci.oraclecloud.com:9092");
	properties.put("security.protocol", "SASL_SSL");
	properties.put("sasl.mechanism", "PLAIN");
	properties.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"{tenancyName}/{username}/{streamPoolId}\" password=\"{authToken}\";");

Paramètres recommandés pour les producteurs de la trousse SDK Java :

properties.put("retries", 5); // retries on transient errors and load balancing disconnection
		properties.put("max.request.size", 1024 * 1024); // limit request size to 1MB

Paramètres recommandés pour les consommateurs de la trousse SDK Java :

properties.put("max.partition.fetch.bytes", 1024 * 1024); // limit request size to 1MB per partition

Pour la trousse SDK Librdkafka

Paramètres recommandés pour la trousse SDK Librdkafka :

'metadata.broker.list': 'streaming.{region}.oci.oraclecloud.com:9092',
			 'security.protocol': 'SASL_SSL',
		 	 'sasl.mechanisms': 'PLAIN',
			 'sasl.username': '{tenancyName}/{username}/{streamPoolID}',
		         'sasl.password': '{authToken}'

Paramètres recommandés pour les producteurs de la trousse SDK Librdkafka :

'message.send.max.retries': 5 // retries on transient errors and load balancing disconnection
		'max.request.size': 1024 * 1024 // limit request size to 1 MB

Paramètres recommandés pour les consommateurs de la trousse SDK Librdkafka :

'max.partition.fetch.bytes': 1024 * 1024 // limit request size to 1 MB per partition

Autorisation du principal d'instance pour la trousse SDK pour Java

Si vous utilisez la trousse SDK pour Java, vous pouvez autoriser une instance à interagir avec le service de diffusion en continu au lieu d'utiliser des jetons d'authentification.

Pour configurer la trousse SDK pour Java pour l'autorisation du principal d'instance :

Vérifiez que vous disposez d'un fichier de configuration de trousse SDK et d'interface de ligne de commande valide pour Oracle Cloud Infrastructure (OCI).
Importez la trousse SDK Oracle Cloud Infrastructure pour Java dans votre projet. Pour plus d'informations, voir Introduction à la trousse SDK pour Java.

Ajoutez la trousse SDK Oracle Cloud Infrastructure suivante pour la dépendance Java :

<dependency>
  <groupId>com.oracle.oci.sdk</groupId>
  <artifactId>oci-apisdk-java-sdk-addons-sasl</artifactId>
  <optional>false</optional>
  <version>1.13.1</version> <!-- that's the minimum version to use -->
</dependency>

Modifiez la propriété sasl.mechanism de votre configuration de client Kafka :

properties.put("sasl.mechanism", OciMechanism.OCI_RSA_SHA256.mechanismName());

Modifiez la propriété sasl.jaas.config de votre configuration de client Kafka à l'aide de l'une des options suivantes :

properties.put("sasl.jaas.config", "com.oracle.bmc.auth.sasl.InstancePrincipalsLoginModule required intent=\"streamPoolId:<streamPoolId>\";");

properties.put("sasl.jaas.config", "com.oracle.bmc.auth.sasl.UserPrincipalsLoginModule required config=\"<pathToConfig>\" profile=\"<profile>\" intent=\"streamPoolId:<streamPoolId>\";");

Si le paramètre config n'est pas spécifié, le chemin de configuration par défaut est utilisé (~/.oci/config).
Si le paramètre profile n'est pas spécifié, le profil par défaut est utilisé (DEFAULT).

Pour plus d'informations

Documentation sur Apache Kafka
Billet de blogue : Producteur et consommateur du service de diffusion en continu Oracle