spring-schema-registry

KO | EN

gitlite — search

#hacktoberfest #ai #llm #golang #go #rust #deep-learning #linux #docker #macos #windows #vue #security #openai

spring-schema-registry

(★ 30)

The purpose of this module is to solve the problem of multiple keystores using Spring Boot + Kafka + Schema Registry + SSL and Spring Schema Resgistry Test

spring-schema-registry 최신버젼 다운로드

최종 버전 다운로드 (.zip)

// repository documentation

Was this content helpful?

★ 0 (0 ratings)

# Spring Schema Registry

[![Build, Publish Snapshot and Pull Request](https://github.com/mvallim/spring-schema-registry/actions/workflows/cd-snapshot.yml/badge.svg)](https://github.com/mvallim/spring-schema-registry/actions/workflows/cd-snapshot.yml)
[![Quality Gate Status](https://sonarcloud.io/api/project_badges/measure?project=spring-schema-registry&metric=alert_status)](https://sonarcloud.io/dashboard?id=spring-schema-registry)
[![Coverage](https://sonarcloud.io/api/project_badges/measure?project=spring-schema-registry&metric=coverage)](https://sonarcloud.io/dashboard?id=spring-schema-registry)
[![Maven Central](https://img.shields.io/maven-central/v/com.github.mvallim/spring-schema-registry)](https://img.shields.io/maven-central/v/com.github.mvallim/spring-schema-registry)
[![Hex.pm](https://img.shields.io/hexpm/l/plug.svg)](http://www.apache.org/licenses/LICENSE-2.0)

The purpose of this module is to solve the problem of multiple keystores using Spring Boot + Kafka + Schema Registry + SSL

## Notice

* Compatible with **JDK 8, 11, 17 and 21**
* Compatible with **schema-registry version 5.3.1 or greater**
* Compatible with **avro version 1.9.1 or greater**
  
## Problem description

1. **When**
    * We have a Spring Boot application exposing SSL end-points with a first distinct certificate;
    * We have communication with Kafka via SSL with a second distinct certificate;
    * We have the communication with Schema Registry with the same certificate used for communication with Kafka or a separate third party certificate;

2. **Scenarios**

| Protocol | Spring Boot | Kafka | Schema Registry | Result |
    |:--------:|:-----------:|:-----:|:---------------:|:------:|
    | SSL      | Yes         | Not   | Not             | **Ok** |
    | SSL      | Yes         | Yes   | Not             | **Ok** |
    | SSL      | Yes         | Yes   | Yes             | Fail   |
    | SSL      | Not         | Yes   | Yes             | **Ok** |
    | SSL      | Not         | Not   | Yes             | **Ok** |
    | SSL      | Not         | Not   | Not             | **Ok** |

The failure happens in a scenario where we would expect it to be fully functional, where the application uses one certificate to securely expose endpoints, and uses other certificates to communicate with Schema Registry and Kafka.

The problem happens because the `kafka-avro-serializer` component uses the JVM variables `javax.net.ssl.trustStore`, `javax.net.ssl.keyStore`, `javax.net.ssl.trustStorePassword` and `javax.net.ssl.keyStorePassword`, and these variables apply to the whole application. As a consequence, if we use a certificate to publish the application API, it will be used by the `kafka-avro-serializer` component.

It is intended that the application uses a certificate to expose its API and use a second certificate to communicate with the **Schema Registry**.

## 1. Quick Start

This chapter will show you how to use Kafka + Schema Registry over SSL.

## 1.1 Create certificates [generate-certificates.sh](./generate-certificates.sh)

Run 
```console
./generate-certificates.sh
```

Expected output:

```console
$./generate-certificates.sh

=> ROOT and CA
 => Generate the private keys (for root and ca)
 => Generate the root certificate
 => Generate certificate for ca signed by root (root -> ca)
 => Import ca cert chain into ca.jks
=> Kafka Server
 => Generate the private keys (for the server)
 => Generate certificate for the server signed by ca (root -> ca -> kafka-server)
 => Import the server cert chain into kafka.server.keystore.jks
 => Import the server cert chain into kafka.server.truststore.jks
=> Schema Registry Server
 => Generate the private keys (for schema-registry-server)
 => Generate certificate for the server signed by ca (root -> ca -> schema-registry-server)
 => Import the server cert chain into schema-registry.server.keystore.jks
 => Import the server cert chain into schema-registry.server.truststore.jks
=> Control Center Server
 => Generate the private keys (for control-center-server)
 => Generate certificate for the server signed by ca (root -> ca -> control-center-server)
 => Import the server cert chain into control-center.server.keystore.jks
 => Import the server cert chain into control-center.server.truststore.jks
=> Appplication Client
 => Generate the private keys (for application-client)
 => Generate certificate for the client signed by ca (root -> ca -> application-client)
 => Import the client cert chain into application.client.keystore.jks
 => Import the client cert chain into application.client.truststore.jks
=> Clean up
 => Move files
```

## 1.2 Running stack [docker-compose-yml](./docker-compose.yml)

Run
```console
docker-compose up -d
```

Expected output:

```console
Creating network "schema" with the default driver
Creating zookeeper ... done
Creating kafka     ... done
Creating schema-registry ... done
Creating control-center  ... done
```

### 1.3 Dependencies

You can pull it from the central Maven repositories:

```xml
<dependency>
  <groupId>com.github.mvallim</groupId>
  <artifactId>spring-schema-registry</artifactId>
  <version>2.1.1</version>
</dependency>
```

If you want to try a snapshot version, add the following repository:

```xml
<repository>
  <id>sonatype-snapshots</id>
  <name>Sonatype Snapshots</name>
  <url>https://oss.sonatype.org/content/repositories/snapshots</url>
  <snapshots>
    <enabled>true</enabled>
  </snapshots>
</repository>
```

#### Gradle

```groovy
implementation 'com.github.mvallim:spring-schema-registry:2.1.1'
```

If you want to try a snapshot version, add the following repository:

```groovy
repositories {
  maven {
    url "https://oss.sonatype.org/content/repositories/snapshots"
  }
}
```

### 1.4 SpecificKafkaAvroDeserializer

This Avro event deserializer is intended to deserialize events from a topic that have events registered in the Schema Registry and **with the Avros imported into the application**.

This deserializer does not handle events that are not registered in Schema Registry and/or Avros with compatibility problem, in these cases an exception will be thrown.

***Attention**: You must use YAML **or** PROPERTIES.*

#### 1.4.1 application.yaml

```yaml
spring:
  kafka:
    bootstrap-servers: localhost:9092
    properties:
      security.protocol: SSL
      auto.register.schemas: true
      value.subject.name.strategy: io.confluent.kafka.serializers.subject.TopicRecordNameStrategy
      schema.registry.url: https://localhost:8082
    ssl:
      protocol: SSL
      key-password: changeit
      key-store-location: file:certificates/application/application.client.keystore.jks
      key-store-password: changeit
      key-store-type: JKS
      trust-store-location: file:certificates/application/application.client.truststore.jks
      trust-store-password: changeit
      trust-store-type: JKS
    consumer:
      properties:
        max.poll.interval.ms: 3000
        specific.avro.reader: true
      group-id: people
      auto-offset-reset: earliest
      enable-auto-commit: true
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.schemaregistry.deserializer.SpecificKafkaAvroDeserializer
    producer:
      acks: all
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.schemaregistry.serializer.SpecificKafkaAvroSerializer

server:
  port: 8443
  ssl:
    key-store: classpath:keystore.jks
    key-store-password: changeit
    keyStoreType: JKS
    trust-store: classpath:truststore.jks
    trust-store-password: changeit
    trustStoreType: JKS
```

#### 1.4.2 application.properties

```properties
spring.kafka.bootstrap-servers=localhost:9092

spring.kafka.properties.auto.register.schemas=false
spring.kafka.properties.value.subject.name.strategy=io.confluent.kafka.serializers.subject.TopicRecordNameStrategy
spring.kafka.properties.schema.registry.url=https://localhost:8082
spring.kafka.properties.security.protocol=SSL
spring.kafka.properties.auto.register.schemas=true
spring.kafka.properties.value.subject.name.strategy=io.confluent.kafka.serializers.subject.TopicRecordNameStrategy

spring.kafka.ssl.protocol=SSL
spring.kafka.ssl.key-password=changeit
spring.kafka.ssl.key-store-location=file:certificates/application/application.client.keystore.jks
spring.kafka.ssl.key-store-password=changeit
spring.kafka.ssl.key-store-type=JKS
spring.kafka.ssl.trust-store-location=file:certificates/application/application.client.truststore.jks
spring.kafka.ssl.trust-store-password=changeit
spring.kafka.ssl.trust-store-type=JKS

spring.kafka.consumer.properties.max.poll.interval.ms=3000
spring.kafka.consumer.properties.specific.avro.reader=true
spring.kafka.consumer.group-id=people
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=true
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.springframework.schemaregistry.deserializer.SpecificKafkaAvroDeserializer

spring.kafka.producer.acks=all
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.springframework.schemaregistry.serializer.SpecificKafkaAvroSerializer

server.port=8443
server.ssl.key-store=classpath:keystore.jks
server.ssl.key-store-password=changeit
server.ssl.keyStoreType=JKS
server.ssl.trust-store=classpath:truststore.jks
server.ssl.trust-store-password=changeit
server.ssl.trustStoreType=JKS
```

### 1.5 GenericKafkaAvroDeserializer

This Avro event deserializer is intended to deserialize each and every event from a topic that is registered in the Schema Registry, **with or without Avros imported into the application**.

For example, currently the application is prepared to consume and deserialize events of type `A`, but if some other event that the application is not prepared to consume occurs in this topic, it will be transformed into a `GenericRecord` (provided that this type of event is registered in the Schema Registry). So it is possible to consume all events without deserialization errors. This prevents the application that wants to consume only certain types of events from a topic, from not forcing the import of all Avros (.svc) into its application.

This deserializer does not handle events that are not registered in Schema Registry and/or Avros with compatibility problem, in these cases an exception will be thrown.

***Attention**: You must use YAML **or** PROPERTIES.*

#### 1.5.1 application.yaml

```yaml
spring:
  kafka:
    bootstrap-servers: localhost:9092
    properties:
      security.protocol: SSL
      auto.register.schemas: true
      value.subject.name.strategy: io.confluent.kafka.serializers.subject.TopicRecordNameStrategy
      schema.registry.url: https://localhost:8082
    ssl:
      protocol: SSL
      key-password: changeit
      key-store-location: file:certificates/application/application.client.keystore.jks
      key-store-password: changeit
      key-store-type: JKS
      trust-store-location: file:certificates/application/application.client.truststore.jks
      trust-store-password: changeit
      trust-store-type: JKS
    consumer:
      properties:
        max.poll.interval.ms: 3000
        specific.avro.reader: true
      group-id: people
      auto-offset-reset: earliest
      enable-auto-commit: true
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.schemaregistry.deserializer.GenericKafkaAvroDeserializer
    producer:
      acks: all
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.schemaregistry.serializer.SpecificKafkaAvroSerializer

server:
  port: 8443
  ssl:
    key-store: classpath:keystore.jks
    key-store-password: changeit
    keyStoreType: JKS
    trust-store: classpath:truststore.jks
    trust-store-password: changeit
    trustStoreType: JKS
```

#### 1.5.2 application.properties

```properties
spring.kafka.bootstrap-servers=localhost:9092

spring.kafka.properties.auto.register.schemas=false
spring.kafka.properties.value.subject.name.strategy=io.confluent.kafka.serializers.subject.TopicRecordNameStrategy
spring.kafka.properties.schema.registry.url=https://localhost:8082
spring.kafka.properties.security.protocol=SSL
spring.kafka.properties.auto.register.schemas=true
spring.kafka.properties.value.subject.name.strategy=io.confluent.kafka.serializers.subject.TopicRecordNameStrategy

spring.kafka.ssl.protocol=SSL
spring.kafka.ssl.key-password=changeit
spring.kafka.ssl.key-store-location=file:certificates/application/application.client.keystore.jks
spring.kafka.ssl.key-store-password=changeit
spring.kafka.ssl.key-store-type=JKS
spring.kafka.ssl.trust-store-location=file:certificates/application/application.client.truststore.jks
spring.kafka.ssl.trust-store-password=changeit
spring.kafka.ssl.trust-store-type=JKS

spring.kafka.consumer.properties.max.poll.interval.ms=3000
spring.kafka.consumer.properties.specific.avro.reader=true
spring.kafka.consumer.group-id=people
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=true
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.springframework.schemaregistry.deserializer.GenericKafkaAvroDeserializer

spring.kafka.producer.acks=all
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.springframework.schemaregistry.serializer.SpecificKafkaAvroSerializer

server.port=8443
server.ssl.key-store=classpath:keystore.jks
server.ssl.key-store-password=changeit
server.ssl.keyStoreType=JKS
server.ssl.trust-store=classpath:truststore.jks
server.ssl.trust-store-password=changeit
server.ssl.trustStoreType=JKS
```

### 1.6 Configure beans

#### 1.6.1 ProducerConfig

```java
@Configuration
public class ProducerConfig {

@Bean
  public ProducerFactory<String, GenericRecord> producerFactory(final KafkaProperties kafkaProperties) {
    return new DefaultKafkaProducerFactory<>(kafkaProperties.buildProducerProperties());
  }

@Bean
  public KafkaTemplate<String, GenericRecord> kafkaTemplate(final ProducerFactory<String, GenericRecord> producerFactory) {
    return new KafkaTemplate<>(producerFactory);
  }
}
```

#### 1.6.2 ConsumerConfig

```java
@EnableKafka
@Configuration
public class ConsumerConfig {

@Bean
  public ConsumerFactory<String, GenericRecord> consumerFactory(final KafkaProperties kafkaProperties) {
    return new DefaultKafkaConsumerFactory<>(kafkaProperties.buildConsumerProperties());
  }

@Bean
  public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, GenericRecord>> containerFactory(final ConsumerFactory<String, GenericRecord> consumerFactory) {
    final ConcurrentKafkaListenerContainerFactory<String, GenericRecord> containerFactory = new ConcurrentKafkaListenerContainerFactory<>();
    containerFactory.setConsumerFactory(consumerFactory);
    containerFactory.setConcurrency(20);
    return containerFactory;
  }
}
```

## Contributing

Please read [CONTRIBUTING.md](CONTRIBUTING.md) for details on our code of conduct, and the process for submitting pull requests to us.

## Versioning

We use [GitHub](https://github.com/mvallim/spring-schema-registry) for versioning. For the versions available, see the [tags on this repository](https://github.com/mvallim/spring-schema-registry/tags).

## Authors

* **Marcos Vallim** - *Founder, Author, Development, Test, Documentation* - [mvallim](https://github.com/mvallim)
* **Carlos Batistão** - *Development, Test, Documentation* - [cezbatistao](https://github.com/cezbatistao)
* **Ricardo Comar** - *Development, Test* - [ricardo-comar](https://github.com/ricardo-comar)

See also the list of [contributors](CONTRIBUTORS.txt) who participated in this project.

## License

This project is licensed under the Apache License - see the [LICENSE](LICENSE) file for details

spring-schema-registry

spring-schema-registry 최신버젼 다운로드

오늘의 추천 프로젝트

Was this content helpful?

Recent Feedback

spring-schema-registry

spring-schema-registry 최신버젼 다운로드

오늘의 추천 프로젝트

Was this content helpful?

Recent Feedback

Download README