Построение гибкой и распределенной архитектуры с использованием Kafka и Kafka Connect: Часть 2 — Получатель и Helm Chart

Данная статья является продолжением статьи — Ивентная модель данных с использованием Kafka и Kafka Connect: Построение гибкой и распределенной архитектуры.

Добро пожаловать во вторую часть статьи о построении гибкой и распределенной архитектуры с использованием Apache Kafka и Kafka Connect! В первой части мы ознакомились с ивентной моделью данных, разработали сервис отправителя и настроили интеграцию с Kafka, чтобы асинхронно отправлять сообщения. Теперь настало время рассмотреть вторую часть этого увлекательного проекта.

В этой части статьи мы сфокусируемся на реализации получателя на Java, который будет получать Avro-схемы из Schema Registry и читать сообщения из Kafka, отправленные сервисом отправителя через Kafka Connect. Получатель будет обрабатывать сообщения из топика, обогащать их своими данными и сохранять результаты в своей базе данных.

Цели статьи:

1. Разработка Java-приложения в качестве получателя данных из Kafka.

2. Изучение работы с Avro-схемами и Schema Registry для обеспечения совместимости сообщений.

3. Обработка сообщений из Kafka с использованием асинхронного подхода.

4. Интеграция с PostgreSQL для сохранения обработанных данных.

5. Подготовка Helm Chart для развертывания отправителя и получателя в Kubernetes.

6. Конфигурирование Kubernetes для обеспечения масштабируемости и отказоустойчивости архитектуры.

Подготовка Java-приложения в качестве получателя

Для начала подготовим наш POM файл

<dependencies>  <dependency>    <groupId>com.google.guava</groupId>    <artifactId>guava</artifactId>    <version>32.1.1-jre</version>  </dependency>  <dependency>    <groupId>io.confluent</groupId>    <artifactId>kafka-avro-serializer</artifactId>    <version>7.4.0</version>  </dependency>  <dependency>    <groupId>io.confluent</groupId>    <artifactId>kafka-schema-registry-client</artifactId>    <version>7.4.0</version>  </dependency>  <dependency>    <groupId>org.springframework.boot</groupId>    <artifactId>spring-boot-starter-data-jpa</artifactId>  </dependency>  <dependency>    <groupId>org.springframework.boot</groupId>    <artifactId>spring-boot-starter-actuator</artifactId>  </dependency>  <dependency>    <groupId>org.springframework.boot</groupId>    <artifactId>spring-boot-starter-web</artifactId>  </dependency>  <dependency>    <groupId>org.springframework.kafka</groupId>    <artifactId>spring-kafka</artifactId>  </dependency>  <dependency>    <groupId>org.postgresql</groupId>    <artifactId>postgresql</artifactId>    <scope>runtime</scope>  </dependency>  <dependency>    <groupId>org.projectlombok</groupId>    <artifactId>lombok</artifactId>    <optional>true</optional>  </dependency>  <dependency>    <groupId>org.springframework.boot</groupId>    <artifactId>spring-boot-starter-test</artifactId>    <scope>test</scope>  </dependency>  <dependency>    <groupId>org.springframework.kafka</groupId>    <artifactId>spring-kafka-test</artifactId>    <scope>test</scope>  </dependency> </dependencies>

Здесь мы добавим библиотеки необходимые для работы со Schema Regestry, PostgreSQL, Kafka и проч. Чтобы успешно выгрузить библиотеки io.confluent необходимо подключиться к их репозиторию

<repositories>  <repository>    <id>confluent</id>    <url>https://packages.confluent.io/maven/</url>  </repository> </repositories>

Теперь напишем yaml конфигурацию для проекта:

# Порт сервера server:   port: 8081 # Настройки Spring spring:   main:     allow-bean-definition-overriding: true   application:     name: kafka-example-consumer   jpa:     # Диалект базы данных     database-platform: org.hibernate.dialect.PostgreSQLDialect     generate-ddl: false     hibernate:       ddl-auto: update     show-sql: true     properties:       hibernate:         generate_statistics: false   datasource:     # URL для подключения к базе данных PostgreSQL     url: jdbc:postgresql://localhost:5432/public     username: postgres     password: postgres   kafka:     schema:       registry:         # Пути до хранилищ схем если у нас их несколько (дополнительная конфигурация)         urls: http://schema-registry-cp-schema-registry:8081     bootstrap-servers: localhost:29092     listener:       # Как подтверждаем получение сообщений       ack-mode: record     producer:       client-id: ${spring.application.name}       # Сериализатор ключа       key-serializer: org.apache.kafka.common.serialization.StringSerializer       # Сериализатор значения в DLT очередь       value-serializer: ru.marmarks.consumer.config.DltMessageSerializer       retries: 3     consumer:       group-id: ${spring.application.name}       autoOffsetReset: earliest       # Сериализаторы для ключа и значения с обработкой ошибок       keyDeserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer       valueDeserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer       properties:         # Тип значения в JSON по умолчанию         spring.json.value.default.type: org.apache.avro.generic.GenericRecord         spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer        # Сериализатор значения         spring.deserializer.value.delegate.class: io.confluent.kafka.serializers.KafkaAvroDeserializer         specific:           avro:             reader: true             # Хранилище схем по умолчанию         schema:           registry:             url: http://schema-registry-cp-schema-registry:8081 # Настройки Kafka kafka:   topics:     # Топик из которого будем получать записи     personal-data: kafka-connect-personal_data   services:     # Для определения того из каких топиков мы поддерживаем чтение     personal-data: kafka-connect-personal_data

Подключение к Schema Registry и Kafka брокеру

Теперь перейдём к конфигурированию Kafka. Комментарии приведены в коде. Здесь мы конфигурируем подключение к Schema Regestry, слушателя и отправку сообщений, которые не вышло обработать в dlt очередь

@Configuration @EnableKafka @RequiredArgsConstructor @EnableConfigurationProperties(SchemaRegistryProperties.class) public class KafkaConfiguration {  private static final int DEFAULT_CACHE_CAPACITY = 200;  private static final String DLT_TOPIC_SUFFIX = ".dlt";  private final ConsumerFactory<String, GenericRecord> consumerFactory;  private final ProducerFactory<Object, Object> producerFactory;   @Bean  public ConcurrentKafkaListenerContainerFactory<String, GenericRecord> objectKafkaListenerContainerFactory(      DefaultErrorHandler errorHandler  ) {    ConcurrentKafkaListenerContainerFactory<String, GenericRecord> factory = new ConcurrentKafkaListenerContainerFactory<>();    factory.setConsumerFactory(consumerFactory);    factory.setCommonErrorHandler(errorHandler);    return factory;  }   /**   * Клиент для кеширования схем   */  @Bean  public CachedSchemaRegistryClient cachedSchemaRegistryClient(      SchemaRegistryProperties schemaRegistryProperties) {    // Сервис для коммуникацией со Schema Registry    RestService restService = new RestService(schemaRegistryProperties.urls());    //Также передаём размер кэша схем    return new CachedSchemaRegistryClient(restService, DEFAULT_CACHE_CAPACITY);  }   @Bean  public KafkaTemplate<Object, Object> kafkaTemplate() {    return new KafkaTemplate<>(producerFactory);  }   /**   * Публикатор в dead-letter topic.   */  @Bean  public DeadLetterPublishingRecoverer publisher(KafkaTemplate<Object, Object> bytesTemplate) {    //  Определяем логику выбора партиции для отправки сообщения в DLT.    //  В данном случае, создаём новый объект TopicPartition, используя имя    //топика (consumerRecord.topic()) и добавляя суффикс DLT_TOPIC_SUFFIX,    //  а также номер партиции (consumerRecord.partition()).    //  Следовательно в DLT топике должно быть столько партиций    //сколько и в топике откуда читаем    return new DeadLetterPublishingRecoverer(bytesTemplate, (consumerRecord, exception) ->        new TopicPartition(consumerRecord.topic() + DLT_TOPIC_SUFFIX, consumerRecord.partition()));  }   /**   * Обработчик исключений при получении сообщений из kafka по умолчанию.   */  @Bean  public DefaultErrorHandler errorHandler(      DeadLetterPublishingRecoverer deadLetterPublishingRecoverer) {    final var handler = new DefaultErrorHandler(deadLetterPublishingRecoverer);    // Обрабатываем любые исключения и отправляем в DLT    handler.addNotRetryableExceptions(Exception.class);    return handler;  }  }

Посмотрим на сущность, которую будем обрабатывать: 

public class ClientData {   @Id  @GeneratedValue(strategy = GenerationType.IDENTITY)  private Long id;  private String bankBic;  private String bankName;  private String fio;  private Long correlationId;  private LocalDateTime lastUpdate;  }

Добавили новые поля, относительно отправляемой сущности — fio, correlationId. Предположим, что мы заполняем ФИО, обращаясь в какой-то сторонний сервис.

Обработка сообщений из топика

Теперь перейдём к тому, как будем обрабатывать приходящее сообщение

public class KafkaConsumerListeners {   // Получим сервисы, которые смогут обрабатывать приходящие сообщения  private final List<ServiceResolver> serviceResolverList;  // Определяем топики из которых будем читать. Их может быть несколько.   //Также определим конфигурацию листнера  @KafkaListener(topics = "#{'${kafka.topics.personal-data}'.split(',')}", containerFactory = "objectKafkaListenerContainerFactory")  // Принимаем абстрактный класс GenericRecord.   //Мы сможем обработать только то сообщение, для которого определена    //схема в Schema Regestry  void readOrganizationConnectorMessages(ConsumerRecord<String, GenericRecord> message) {    readMessage(message);  }   private void readMessage(ConsumerRecord<String, GenericRecord> message) {    String topic = message.topic();    long offset = message.offset();    log.info("Сообщение из топика: {} offset: {}", topic, offset);     // Получаем само сообщение    GenericRecord value = message.value();     if (value == null) {      log.info("Пустое сообщение из топика: {} offset: {}", topic, offset);      return;    }     // Находим сервис, который сможет обработать сообщение    ServiceResolver resolver = serviceResolverList.stream()        .filter(it -> it.isSupported(topic))        .findFirst()        .orElseThrow(() -> new RuntimeException("Сервис не найден для топика: " + topic));     // Передаём сообщение на обработку    resolver.process(value);     log.info("Сообщение обработано: топик {} offset {}", topic, offset);  }  }

Комментарии по коду приведены непосредственно в коде. Реализация заточена на то, что мы сможем обрабатывать несколько топиков с различными сообщения. Схема каждого сообщения должна быть приведена в Schema Regestry

Работа с Avro-схемами

Посмотрим на обработку сообщения в сервисе

public void process(GenericRecord value) {  Long id = Long.valueOf(String.valueOf(value.get("id")));  String bankBic = String.valueOf(value.get("bank_bic"));  String bankName = String.valueOf(value.get("bank_name"));  String lastUpdate = String.valueOf(value.get("last_update"));  long millis = Long.parseLong(lastUpdate);  Instant instant = Instant.ofEpochMilli(millis);  LocalDateTime lastUpdateTime = instant.atZone(ZoneId.systemDefault()).toLocalDateTime();   // Подготовим новую сущность на основе приходящих данных  ClientData clientData = new ClientData();  clientData.setBankBic(bankBic);  clientData.setFio("Рауль " + id);  clientData.setBankName(bankName);  clientData.setCorrelationId(id);  clientData.setLastUpdate(lastUpdateTime);   ClientData save = clientDataRepository.save(clientData);  log.info("Данные клиента сохранены: {}", save);  }

В данном случае мы примитивным способом вытаскиваем все поля из сообщения и на основе них создаём новую сущность, которую модифицируем новыми данными, которые мы можем получить обратившись куда-либо ещё или иным способом и записываем в базу

Полный код получателя доступен на GitHub.

Helm Chart

Теперь перейдём к Helm чарту, который будет разворачивать нашего отправителя и получателя.

Подготовим файл с информацией о чарте:

# Версия API Helm Chart apiVersion: v2 # Имя чарта name: marmarks-chart # Версия чарта version: 1.0.0 # Описание description: Some project # Тип type: application sources:   - https://github.com/marmarks maintainers:   - name: Maksim Krylov     email: kryloumaksim@gmail.com     url: https://habr.com/ru/users/Marmaksmark/     # Версия приложения appVersion: 1.0.0

Конфигурации и шаблоны

Следующим шагом подготовим файл values.yaml, который предоставляет конфигурации для чарта. Помимо того, что настройки будут определены в этом файле мы сможем также их переопределять используя --set при запуске чарта

producer:   # Конфигурация деплоймента   deployment:     # Имя деплоймента     name: "producer"     # Число реплик     replicas: 1     # Настройки контейнера   container:       name: "producer-container"       resources: { }     # Настройки сервиса   service:       name: "producer-service"       port: 8080     # Информация об образе сервиса   image:       name: marmarks/producer       tag: "0.2"       # Загрузим образ, если локально его нет       pullPolicy: IfNotPresent   consumer:   deployment:     name: "consumer"     replicas: 1   container:     name: "consumer-container"     resources: { }   service:     name: "consumer-service"     port: 80     #  тип сервиса: ClusterIP, NodePort, LoadBalancer     type: LoadBalancer   image:     name: marmarks/consumer     tag: "0.5"     pullPolicy: IfNotPresent

Также мы определим файл _helpers.tpl содержащий вспомогательные шаблоны, такие как:

{{- define "marmarks-chart.selectorLabels" -}} app.kubernetes.io/name: {{ .Chart.Name }} app.kubernetes.io/instance: {{ .Release.Name }} {{- end -}}

{{- define "marmarks-chart.chart" -}} {{- printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" | trunc 60 | trimSuffix "-" -}} {{- end -}}

{{- define "marmarks-chart.labels" -}} helm.sh/chart: {{ include "marmarks-chart.chart" . }} {{ include "marmarks-chart.selectorLabels" . }} {{- if .Chart.AppVersion }} app.kubernetes.io/version: {{ .Chart.AppVersion | quote }} {{- end }} app.kubernetes.io/managed-by: {{ .Release.Service }} {{- end -}}

{{- define "marmarks-chart.propertiesHash" -}} {{- $producerSecrets := include (print $.Template.BasePath "/producer-secrets.yaml") . | sha256sum -}} {{- $consumerSecrets := include (print $.Template.BasePath "/consumer-secrets.yaml") . | sha256sum -}} {{- $producerConfig := include (print $.Template.BasePath "/producer-config.yaml") . | sha256sum -}} {{- $consumerConfig := include (print $.Template.BasePath "/consumer-config.yaml") . | sha256sum -}} {{ print $producerSecrets $producerConfig $consumerSecrets $consumerConfig | sha256sum }} {{- end -}}

{{- define "marmarks-chart.producer.defaultName" -}} {{- printf "producer-%s" .Release.Name -}} {{- end -}}  {{- define "marmarks-chart.producer.deployment.name" -}} {{- default (include "marmarks-chart.producer.defaultName" .) .Values.producer.deployment.name | trunc 63 | trimSuffix "-" -}} {{- end -}}  {{- define "marmarks-chart.producer.container.name" -}} {{- default (include "marmarks-chart.producer.defaultName" .) .Values.producer.container.name | trunc 63 | trimSuffix "-" -}} {{- end -}}  {{- define "marmarks-chart.producer.service.name" -}} {{- default (include "marmarks-chart.producer.defaultName" .) .Values.producer.service.name | trunc 63 | trimSuffix "-" -}} {{- end -}}

{{- define "marmarks-chart.consumer.defaultName" -}} {{- printf "consumer-%s" .Release.Name -}} {{- end -}}  {{- define "marmarks-chart.consumer.deployment.name" -}} {{- default (include "marmarks-chart.consumer.defaultName" .) .Values.consumer.deployment.name | trunc 63 | trimSuffix "-" -}} {{- end -}}  {{- define "marmarks-chart.consumer.container.name" -}} {{- default (include "marmarks-chart.consumer.defaultName" .) .Values.consumer.container.name | trunc 63 | trimSuffix "-" -}} {{- end -}}  {{- define "marmarks-chart.consumer.service.name" -}} {{- default (include "marmarks-chart.consumer.defaultName" .) .Values.consumer.service.name | trunc 63 | trimSuffix "-" -}} {{- end -}}

{{- define "marmarks-chart.producerSecrets.defaultName" -}} {{- printf "producer-secrets-%s" .Release.Name -}} {{- end -}}  {{- define "marmarks-chart.consumerSecrets.defaultName" -}} {{- printf "consumer-secrets-%s" .Release.Name -}} {{- end -}}  {{- define "marmarks-chart.producerConfig.defaultName" -}} {{- printf "producer-config-%s" .Release.Name -}} {{- end -}}  {{- define "marmarks-chart.consumerConfig.defaultName" -}} {{- printf "consumer-config-%s" .Release.Name -}} {{- end -}}

Получатель

Теперь, когда конфигурации и шаблоны описаны мы можем описать конфигурационный файл consumer-config.yaml для отправителя:

apiVersion: v1 kind: ConfigMap metadata:   name: {{ include "marmarks-chart.consumerConfig.defaultName" . }}   labels:     {{- include "marmarks-chart.labels" . | nindent 4 }} data:   SPRING_PROFILES_ACTIVE: "dev"   SERVER_PORT: "8080"   SPRING_KAFKA_BOOTSTRAP_SERVERS: "kafka:9092"   SPRING_DATASOURCE_URL: "jdbc:postgresql://postgresql-dev:5432/postgres"

Здесь мы определили на каком порту будем запускать наше приложение, выбранный профиль, url брокера и url до базы данных. Логин-пароль для аутентификации будем хранить в секретах consumer-secrets.yaml:

# Манифест для создания секретов #  Версия API apiVersion: v1 # Тип манифеста kind: Secret # Метаданные манифеста metadata:   # Имя манифеста   name: {{ include "marmarks-chart.consumerSecrets.defaultName" . }}   labels:     # Метки манифеста     {{- include "marmarks-chart.labels" . | nindent 4 }}   # Тип секрета: неструктурированные данные. Храни просто пару "ключ-значение"   # Без автоматического обновления, управления, ротации и т.д. type: Opaque stringData:   SPRING_DATASOURCE_USERNAME: "postgres"   SPRING_DATASOURCE_PASSWORD: "pgpass"

Подробнее про секреты и их типы в статье — Безопасное хранение secrets в Kubernetes

Осталось описать лишь дейломент и сервис получателя consumer.yaml. Исчерпывающие комментарии оставлены в коде:

apiVersion: apps/v1 kind: Deployment metadata:   name: {{ include "marmarks-chart.consumer.deployment.name" . }}   labels:     {{- include "marmarks-chart.labels" . | nindent 4 }}     tier: consumer {{/*    Спецификация развёртывания*/}} spec: {{/*  Число реплик*/}}   replicas: {{ .Values.consumer.deployment.replicas }}   selector: {{/*    Метки подов*/}}     matchLabels:       {{- include "marmarks-chart.selectorLabels" . | nindent 6 }}       tier: consumer {{/*      Стратегия обновления развёртывания*/}}   strategy: {{/*    Пошаговое обновление*/}}     rollingUpdate: {{/*      Максимальное количество реплик для обновления*/}}       maxSurge: 25% {{/*      Максимальное число реплик недоступных во время обновления*/}}       maxUnavailable: 25% {{/*      Тип стратегии развёртывания*/}}     type: RollingUpdate {{/*    Шаблон создания подов*/}}   template:     metadata: {{/*      Аннотации*/}}       annotations: {{/*        Проверка чек суммы для обновления секретов и конфигов*/}}         checksum/config: {{ include "marmarks-chart.propertiesHash" . }}       labels:         {{- include "marmarks-chart.selectorLabels" . | nindent 8 }}         tier: consumer {{/*        Спецификации пода*/}}     spec: {{/*      Контейнер, который будет запущен в поде*/}}       containers:         - name: {{ include "marmarks-chart.consumer.container.name" . }}           image: "{{ .Values.consumer.image.name }}:{{ .Values.consumer.image.tag }}"           imagePullPolicy: {{ .Values.consumer.image.pullPolicy }} {{/*          Переменные окружения из конфиг мапы*/}}           envFrom:             - configMapRef:                 name: {{ include "marmarks-chart.consumerConfig.defaultName" . }} {{/*          Переменные окружения секреты*/}}             - secretRef:                 name: {{ include "marmarks-chart.producerSecrets.defaultName" . }}           readinessProbe: {{/*            Проверка доступности пода*/}}             httpGet:               path: /actuator/health               port: 8080             initialDelaySeconds: 5             periodSeconds: 3           ports: {{/*            Порты пода*/}}             - containerPort: 8080 {{/*              Протокол*/}}               protocol: TCP           resources: {{/*            Ресурсы выделенные для контейнера*/}}             {{- toYaml .Values.consumer.container.resources | nindent 12 }}  ---  apiVersion: v1 kind: Service metadata:   name: {{ include "marmarks-chart.consumer.service.name" . }}   labels:     {{- include "marmarks-chart.labels" . | nindent 4 }}     tier: consumer spec:   type: {{ .Values.consumer.service.type }}   ports: {{/*    Порт службы*/}}     - port: {{ .Values.consumer.service.port }}       protocol: TCP {{/*      Порт контейнера, на который будет направлен трафик*/}}       targetPort: 8080 {{/*      Имя порта*/}}       name: http {{/*    Селектор для выбора подов, которые будут обслужены сервисом*/}}   selector:     {{- include "marmarks-chart.selectorLabels" . | nindent 4 }}     tier: consumer

Мы определили стратегию развертывания, health checks и прочее. Наш получатель уже полностью готов к приёму трафика из брокера, а брокер уже готов его нам поставлять. Однако мы сделаем аналогичную конфигурацию для отправителя

Отправитель

apiVersion: v1 kind: ConfigMap metadata:   name: {{ include "marmarks-chart.producerConfig.defaultName" . }}   labels:     {{- include "marmarks-chart.labels" . | nindent 4 }} data:   SERVER_PORT: "8080"   SPRING_PROFILES_ACTIVE: "dev"   SPRING_DATASOURCE_URL: "jdbc:postgresql://postgresql-dev:5432/postgres"

# Манифест для создания секретов #  Версия API apiVersion: v1 # Тип манифеста kind: Secret # Метаданные манифеста metadata:   # Имя манифеста   name: {{ include "marmarks-chart.producerSecrets.defaultName" . }}   labels:     # Метки манифеста     {{- include "marmarks-chart.labels" . | nindent 4 }}   # Тип секрета: неструктурированные данные. Храни просто пару  "ключ-значение"   # Без автоматического обновления, управления, ротации и т.д. type: Opaque stringData:   SPRING_DATASOURCE_USERNAME: "postgres"   SPRING_DATASOURCE_PASSWORD: "pgpass"

apiVersion: apps/v1 kind: Deployment metadata:   name: {{ include "marmarks-chart.producer.deployment.name" . }}   labels:     {{- include "marmarks-chart.labels" . | nindent 4 }}     tier: producer spec:   replicas: {{ .Values.producer.deployment.replicas }}   selector:     matchLabels:       {{- include "marmarks-chart.selectorLabels" . | nindent 6 }}       tier: producer   strategy:     rollingUpdate:       maxSurge: 25%       maxUnavailable: 25%     type: RollingUpdate   template:     metadata:       labels:         {{- include "marmarks-chart.selectorLabels" . | nindent 8 }}         tier: producer     spec:       containers:         - name: {{ include "marmarks-chart.producer.container.name" . }}           image: "{{ .Values.producer.image.name }}:{{ .Values.producer.image.tag }}"           imagePullPolicy: {{ .Values.producer.image.pullPolicy }}           envFrom:             - configMapRef:                 name: {{ include "marmarks-chart.producerConfig.defaultName" . }}             - secretRef:                 name: {{ include "marmarks-chart.producerSecrets.defaultName" . }}           ports:             - containerPort: 8080               protocol: TCP           readinessProbe:             httpGet:               path: /actuator/health               port: 8080             initialDelaySeconds: 5             periodSeconds: 3           resources:             {{- toYaml .Values.producer.container.resources | nindent 12 }}  ---  apiVersion: v1 kind: Service metadata:   name: {{ include "marmarks-chart.producer.service.name" . }}   labels:     {{- include "marmarks-chart.labels" . | nindent 4 }}     tier: producer spec:   ports:     - port: {{ .Values.producer.service.port }}       protocol: TCP       targetPort: 8080       name: http   selector:     {{- include "marmarks-chart.selectorLabels" . | nindent 4 }}     tier: producer

Запуск и тестирование архитектуры

Необходимо иметь полностью развёрнутую среду из первой части статьи.

Теперь когда Helm Chart готов к запуску, находясь в директории чарта, выполним команду:

helm install <имя чарта> .

Наш чарт успешно запустится, если логи не скажут иного, и будет доступен. Зайдя в логи пода отправителя мы увидим, что он записал несколько новых сообщений в свою базу. После этого, зайдя в логи получателя, мы увидим, что он прочитал и обработал все сообщения, что были в топике раньше и появились только что. Это происходит весьма быстро.

Проверка работоспособности получателя и отправител

Если в ходе обработки сообщений произошла ошибка, то сообщения, чья обработка была безуспешна отправятся в DLT топик. Для того, чтобы убедиться в успешной записи сообщений, мы зайдём в под PostgreSQL и выполним несколько команд. Для начала авторизуемся:

psql -h postgresql-dev -p 5432 -U postgres -d postgres

Нас попросят ввести пароль, в моём случае это pgpass. После его ввода и успешной авторизации выполним:

SELECT * FROM client_data;

id | bank_bic | bank_name | correlation_id | fio | last_update
—+——————-+——————+———————+————+————————————
1 | 1179896397 | null | 1 | Рауль 1| 2023-08-06 16:08:16.156
2 | -1471064429 | null | 2 | Рауль 2| 2023-08-06 16:08:16.221
3 | -604766198 | null | 3 | Рауль 3| 2023-08-06 16:08:16.228
4 | -771359675 | null | 4 | Рауль 4| 2023-08-06 16:08:16.232
5 | 1082682564 | null | 5 | Рауль 5| 2023-08-06 16:08:16.237
———————————————————————————————————————

Заключение

Если данные будут успешно обработаны, то записи будут получены при выполнении SQL запроса. Таким образом мы видим, что ивентная модель работает и успешно обрабатывает данные в асинхронном режиме, чем обеспечивает Evential Consistency.
Исходный код получателя и Helm Chart доступен на GitHub.