Documentation Oracle Cloud Infrastructure

Intégrer le registre de schémas avec Apache Kafka

Le registre de schémas stocke un historique avec numéro de version de tous les schémas en fonction du nom de sujet indiqué, fournit de nombreux paramètres de compatibilité et permet l'évolution du schéma au fil du temps. Schema Registry vit à l'extérieur et séparément des courtiers Kafka.

Le registre de schémas comporte trois composants principaux :

  • Serveur Web de registre : Application Web exposant les adresses REST utilisées pour gérer les entités de schéma. Utilisez un proxy Web et un équilibreur de charge avec de nombreux serveurs Web pour assurer la haute disponibilité et l'évolutivité. Pour activer Schema Registry HA, reportez-vous à la section Schema Registry HA.
  • Stockage des métadonnées du schéma/stockage pluggable : stockage relationnel contenant les métadonnées des entités de schéma. Le stockage en mémoire et les bases de données MySQL sont pris en charge.
  • Stockage Serdes : stockage de fichiers pour les bocaux sérialiseurs et désérialiseurs. Les systèmes de fichiers locaux et le stockage HDFS sont pris en charge. Le stockage local du système de fichiers est la valeur par défaut.

Les principaux composants de Schema Registry sont les suivants :

  • Serveur Web de registre : Application Web exposant les adresses REST utilisées pour gérer les entités de schéma. Utilisez un proxy Web et un équilibreur de charge avec de nombreux serveurs Web pour assurer la haute disponibilité et l'évolutivité. Reportez-vous à la section Schema Registry HA.
  • Stockage des métadonnées du schéma de stockage pluggable : banque relationnelle qui contient les métadonnées des entités de schéma. Le stockage en mémoire et les bases de données MySQL sont pris en charge.
  • Stockage Serdes : stockage de fichiers pour les bocaux sérialiseurs et désérialiseurs. Les systèmes de fichiers locaux et le stockage HDFS sont pris en charge. Le stockage local du système de fichiers est la valeur par défaut.

Interagir avec le registre de schémas

Utilisez les clients Schema Registry d'une application pour interagir directement avec le serveur Schema Registry afin d'accéder au schéma ou de le modifier.
Remarque

La configuration JAAS est requise pour un cluster compatible Kerberos (cluster hautement disponible) uniquement.
  1. Créez un fichier de configuration JAAS avec le client Registry : 
    RegistryClient {
  com.sun.security.auth.module.Krb5LoginModule required
  useKeyTab=true
  keyTab="/path/to/registry/keytab"
  storeKey=true
  useTicketCache=false
  principal="principal@realm";
};
  2. Dans le code Java, ajoutez ce qui suit à la configuration de l'émetteur Kafka : 
    import com.hortonworks.registries.schemaregistry.serdes.avro.kafka.KafkaAvroSerializer;
..
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.setProperty("schema.registry.url", "http://<host name>:9093/api/v1");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); 
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName());
  3. Créez le client Schema Registry en définissant la variable d'environnement avec les configurations client comme suit : 
    System.setProperty("java.security.auth.login.config","/path/to/registryClient_jaas.conf");

Map<String, Object> config = new HashMap<>();
config.put("schema.registry.url", "http://<registry host name>:9093/api/v1");

String topicName = "my-topic";
SchemaRegistryClient schemaRegistryClient = new SchemaRegistryClient(config);

try {
    SchemaMetadata schemaMetadata = new SchemaMetadata.Builder(topicName)
        .type(AvroSchemaProvider.TYPE)
        .schemaGroup("sample-group")
        .description("Sample schema")
        .compatibility(SchemaCompatibility.BACKWARD)
        .build();
    SchemaIdVersion v1 = schemaRegistryClient.addSchemaVersion(schemaMetadata, new SchemaVersion(schemaMetadata, "Initial version of the schema"));
    LOG.info("Registered schema [{}] and returned version [{}]", schemaMetadata, v1);
} catch (SchemaNotFoundException e) {
    LOG.info("Schema addition failed for topic {}", topicName);
    throw e;
}}