- Добро пожаловать в репозиторий проекта автоматизированного тестирования с Kafka
- 1. Что такое Apache Kafka?
- 2. Что такое ZooKeeper?
- 3. Что такое Топик
- 4. Настройка окружения
- 5. Структура проекта
- 6. Работа с Avro
- 7. Тестирование
- 8. Confluent Control Center
- 9. Ссылки
- Этот репозиторий создан для помощи в автоматизации тестирования потоковой передачи событий с Kafka.
- Проект включает в себя настройку Kafka, ZooKeeper, Schema Registry, а также примеры Producers и Consumers на Java с использованием Avro.
- В проекте реализованы интеграционные тесты для проверки функциональности Kafka.
- Многие тексты основаны на официальной документации и переведены для удобства.
Стриминг событий — это цифровой эквивалент центральной нервной системы человеческого тела. Это технологическая основа для постоянно активного мира, где компании всё чаще определяются и автоматизируются посредством программного обеспечения, а «пользователь» в таких системах — это зачастую другое программное обеспечение.
Технически стриминг событий — это практика захвата данных в реальном времени из источников событий (базы данных, датчики, мобильные устройства, облачные сервисы, приложения) в виде потоков событий; надёжное долговременное хранение этих потоков для последующего доступа; обработка, анализ и реакция на эти события как в реальном времени, так и ретроспективно; а также маршрутизация потоков событий к различным целевым системам по необходимости.
Таким образом, стриминг событий обеспечивает непрерывный поток и интерпретацию данных, чтобы нужная информация оказалась в нужном месте в нужное время.
Стриминг событий применяется в широчайшем спектре сфер и организаций.
- Обработка платежей и финансовых транзакций в реальном времени (фондовые биржи, банки, страхование).
- Отслеживание и мониторинг автомобилей, грузовиков, автопарков и грузов в реальном времени (логистика, автомобильная промышленность).
- Непрерывный сбор и анализ данных с датчиков IoT-устройств или другого оборудования (заводы, ветряные электростанции).
- Немедленная реакция на действия и запросы клиентов (розничная торговля, отели, туризм, мобильные приложения).
- Мониторинг состояния пациентов в больницах и прогнозирование изменений для своевременного оказания неотложной помощи.
- Объединение, хранение и предоставление данных от различных подразделений компании.
- Формирование основы для платформ данных, архитектур, ориентированных на события, и микросервисов.
Kafka сочетает в себе три ключевые возможности, позволяя реализовать ваши сценарии стриминга событий "от начала до конца" в едином, проверенном решении:
- Публикация (запись) и подписка (чтение) на потоки событий, включая непрерывный импорт/экспорт ваших данных из других систем.
- Долговременное и надежное хранение потоков событий в течение необходимого вам времени.
- Обработка потоков событий по мере их возникновения или ретроспективно.
Вся эта функциональность предоставляется в распределённом, масштабируемом, эластичном, отказоустойчивом и безопасном виде. Kafka можно развернуть на физическом оборудовании, виртуальных машинах или в контейнерах, локально или в облаке. Вы можете самостоятельно управлять вашей средой Kafka или использовать полностью управляемые сервисы от разных поставщиков.
Kafka — это распределенная система, состоящая из серверов и клиентов, которые общаются по высокопроизводительному сетевому протоколу TCP. Она может быть развернута на физическом оборудовании, виртуальных машинах и контейнерах локально, а также в облачных средах.
Серверы: Kafka запускается как кластер из одного или нескольких серверов, которые могут охватывать несколько дата-центров или облачных регионов. Некоторые из этих серверов образуют слой хранения, называемый брокерами. Другие серверы запускают Kafka Connect для непрерывного импорта и экспорта данных в виде потоков событий, чтобы интегрировать Kafka с существующими системами, такими как реляционные базы данных, а также с другими кластерами Kafka. Чтобы вы могли реализовать критически важные случаи использования, кластер Kafka является высокомасштабируемым и отказоустойчивым: если какой-либо из ваших серверов выйдет из строя, остальные серверы возьмут на себя его работу, чтобы обеспечить непрерывные операции без какой-либо потери данных.
Клиенты: они позволяют вам писать распределенные приложения и микросервисы, которые читают, записывают и обрабатывают потоки событий параллельно, в масштабе и отказоустойчиво, даже в случае сетевых или аппаратных проблем. Kafka поставляется с некоторыми из этих клиентов, которые дополняются десятками клиентов, предоставляемых сообществом Kafka: клиенты доступны для Java и Scala, включая библиотеку верхнего уровня Kafka Streams, для Go, Python, C/C++ и многих других языков программирования, а также REST API.
Zookeeper — это централизованный сервис для хранения конфигурационной информации, служебных имён, обеспечения распределённой синхронизации и групповых сервисов. Все эти типы сервисов в той или иной форме используются распределёнными приложениями. Каждый раз при их реализации возникает множество проблем и ошибок, связанных с гонками и условиями, трудностями в обслуживании. Из-за сложности реализации таких сервисов приложения часто упрощают их, что делает их менее надёжными при изменениях и трудными в управлении. Даже при корректной реализации различные воплощения этих сервисов усложняют управление при развёртывании приложений.
Подробнее о ZooKeeper можно узнать в Wiki Zookeeper.
События организованы и долговременно хранятся в топиках. Упрощённо говоря, топик похож на папку в файловой системе, а события — на файлы в этой папке. Примером названия топика может быть "payments" ("платежи"). Топики в Kafka всегда поддерживают множественность продюсеров и подписчиков: у топика может быть от нуля до нескольких продюсеров, записывающих в него события, и от нуля до нескольких консъюмеров, подписывающихся на эти события. События в топике могут быть прочитаны столько раз, сколько необходимо. В отличие от традиционных систем обмена сообщениями, события не удаляются после чтения. Вместо этого вы задаёте время хранения событий в топике с помощью конфигурации, после чего старые события будут удалены. Производительность Kafka практически не зависит от объёма хранимых данных, поэтому хранить данные долгое время вполне допустимо.
Топики разбиваются на партиции, то есть один топик "расщепляется" на несколько "хранилищ", расположенных на разных брокерах Kafka. Это распределение данных — ключ к масштабированию, позволяющему клиентским приложениям читать и записывать данные параллельно с/на несколько брокеров. Когда новое событие публикуется в топик, оно фактически добавляется в одну из партиций. События с одинаковым ключом (например, идентификатор клиента или автомобиля) записываются в одну и ту же партицию, при этом Kafka гарантирует, что консъюмер, читающий эту партию, получит события в том порядке, в котором они были записаны.
Чтобы сделать данные отказоустойчивыми и высокодоступными, топики могут реплицироваться даже между географическими регионами или дата-центрами. Таким образом, всегда есть несколько брокеров с копиями данных на случай сбоев или необходимости технического обслуживания. Типичная производственная конфигурация — коэффициент репликации равный 3, т.е. три копии ваших данных. Репликация выполняется на уровне партиций топика.
Скачать Kafka можно с официального сайта.
После скачивания распакуйте архив в удобную папку.
Ссылка для скачивания Zookeeper
После этого вы можете распаковать его в любую папку (В моём случае я распаковал в папку C:)
Распакуйте архив в удобную папку.
Здесь вы можете выбрать, запускать ли Kafka и Zookeeper на вашей машине или использовать docker-compose.
-
Перейдите в папку
kafka/bin/. -
Запустите ZooKeeper:
zookeeper-server-start.sh config/zookeeper.properties
-
В новом терминале запустите Kafka:
kafka-server-start.sh config/server.properties
-
Перейдите в папку
kafka/bin/windows. -
Запустите ZooKeeper:
zookeeper-server-start.bat C:\kafka\config\zookeeper.properties
-
В новом терминале запустите Kafka:
kafka-server-start.bat C:\kafka\config\server.properties
На Windows:
kafka-topics.bat --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic topic_userНа Mac/Linux:
kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic topic_userВ проекте уже имеется docker-compose.yml, который запускает Kafka, ZooKeeper, Schema Registry и Control Center.
-
Убедитесь, что Docker установлен на вашем компьютере. Инструкции по установке здесь.
-
Запустите сервисы:
docker-compose up -d
-
Остановить сервисы:
docker-compose down
-
Создание топика (если он не создан автоматически):
docker-compose exec kafka kafka-topics --create --if-not-exists --zookeeper zookeeper:2181 --partitions 1 --replication-factor 1 --topic topic_user
├── docker-compose.yml # Файл для запуска сервисов Kafka, Zookeeper и Schema Registry с помощью Docker
├── pom.xml # Основной файл конфигурации Maven с зависимостями и плагинами
├── src/ # Исходный код проекта
│ ├── main/
│ │ ├── java/ # Основной исходный код Java
│ │ │ └── com/kafka/
│ │ │ ├── consumer/ # Пакет с классами для потребления сообщений из Kafka
│ │ │ │ ├── UserConsumer.java # Потребитель сообщений с типом String
│ │ │ │ └── UserAvroConsumer.java # Потребитель сообщений с типом Avro
│ │ │ ├── model/ # Пакет с моделями данных
│ │ │ │ └── User.java # Модель для пользователя
│ │ │ ├── producer/ # Пакет с классами для отправки сообщений в Kafka
│ │ │ │ ├── UserProducer.java # Продюсер для сообщений с типом String
│ │ │ │ └── UserAvroProducer.java # Продюсер для сообщений с типом Avro
│ │ │ └── utils/ # Пакет утилит для работы с Kafka и другими компонентами
│ │ │ ├── DefaultProperties.java # Класс для получения стандартных свойств Kafka
│ │ │ └── ReadYml.java # Утилита для чтения YAML-файлов
│ ├── resources/ # Ресурсы проекта (например, конфигурационные файлы)
│ │ ├── data/
│ │ │ └── data.yml # Пример YAML-файла с конфигурацией данных
│ │ └── avro/ # Папка с Avro-схемами
│ │ └── user.avsc # Avro-схема для сообщений о пользователе
│ └── test/ # Тесты проекта
│ └── java/
│ └── com/kafka/tests/ # Пакет с тестами для проекта
│ └── KafkaTests.java # Интеграционные тесты для проверки взаимодействия с Kafka
└── README.md # Файл документации проекта
Проект состоит из следующих пакетов:
- com.kafka.consumer: содержит классы
UserConsumerиUserAvroConsumerдля потребления сообщений из Kafka. - com.kafka.producer: содержит классы
UserProducerиUserAvroProducerдля отправки сообщений в Kafka. - com.kafka.model: содержит модель
Userдля стандартных сообщений. - com.kafka.utils: содержит утилитные классы
DefaultPropertiesиReadYmlдля настройки и чтения конфигураций. - com.kafka.tests: содержит интеграционные тесты для проверки функциональности Kafka.
-
Consumer:
UserConsumer: потребляет стандартные JSON-сообщения из топика.UserAvroConsumer: потребляет Avro-сообщения из топика.
-
Producer:
UserProducer: отправляет стандартные JSON-сообщения в топик.UserAvroProducer: отправляет Avro-сообщения в топик.
-
Model:
User: модель пользователя с полямиname,email,age.
-
Utils:
DefaultProperties: содержит методы для получения конфигураций Kafka Producer и Consumer.ReadYml: утилита для чтения YAML-файлов и преобразования их вProperties.
-
Tests:
KafkaTests: интеграционные тесты для проверки работы Producers и Consumers.
Файл pom.xml содержит все необходимые зависимости и плагины для работы проекта, включая Kafka, Avro, Lombok, JUnit и
другие.
Файл docker-compose.yml настроен для запуска следующих сервисов:
- ZooKeeper
- Kafka
- Schema Registry
- Confluent Control Center
Avro — это открытая система сериализации данных, помогающая при обмене данными между системами, языками программирования и фреймворками обработки. Avro определяет бинарный формат для ваших данных и позволяет связать их со страной программирования по вашему выбору.
Kafka работает с любым форматом данных, но для Avro есть специальные возможности из-за его популярности.
Avro похож на JSON по модели данных, но может представляться в JSON или компактном бинарном формате. Он обладает богатым языком описания схем, что упрощает определение и эволюцию формата данных.
Преимущества Avro:
- Прямое отображение из/в JSON.
- Очень компактный формат по сравнению с JSON, где каждая запись повторяет имена полей.
- Высокая скорость.
- Хорошая поддержка множества языков программирования. Можно генерировать Java-объекты для удобной работы с данными событий, но при этом нет обязательной генерации кода — инструменты могут быть написаны универсально для любых потоков данных.
- Богатый расширяемый язык схем в виде простого JSON.
- Чёткое понятие совместимости схем для эволюции данных во времени.
- Управление метаданными облегчает поддержание данных высокого качества, пригодных для масштабной обработки на уровне организации.
Схема Avro для UserAvro:
{
"type": "record",
"name": "UserAvro",
"namespace": "modelAvro.user",
"fields": [
{
"name": "name",
"type": "string"
},
{
"name": "email",
"type": "string"
},
{
"name": "age",
"type": "string"
}
]
}- В IntelliJ IDEA:
- Перейдите в
File > Settings > Plugins. - Найдите и установите "Apache Avro IDL Schema Support".
- Перейдите в
В pom.xml уже добавлены зависимости для Avro и соответствующие плагины.
<dependency>
<groupId>org.apache.avro</groupId>
<artifactId>avro</artifactId>
<version>${avro.version}</version>
</dependency>
<dependency>
<groupId>io.confluent</groupId>
<artifactId>kafka-avro-serializer</artifactId>
<version>${avro.serializer.version}</version>
</dependency>
<plugin>
<groupId>org.apache.avro</groupId>
<artifactId>avro-maven-plugin</artifactId>
<version>1.8.2</version>
<executions>
<execution>
<id>schemas</id>
<phase>generate-sources</phase>
<goals>
<goal>schema</goal>
<goal>protocol</goal>
<goal>idl-protocol</goal>
</goals>
<configuration>
<sourceDirectory>${project.basedir}/src/main/resources/avro/</sourceDirectory>
<outputDirectory>${project.basedir}/src/main/java/</outputDirectory>
</configuration>
</execution>
</executions>
</plugin>Примечание: Убедитесь, что пути <sourceDirectory> и <outputDirectory> соответствуют структуре вашего проекта.
При запуске команды mvn clean install Avro Maven Plugin генерирует классы на основе схемы Avro.
Используйте Builder для создания объектов Avro:
UserAvro avroMessage = UserAvro.newBuilder()
.setName("ivan")
.setEmail("[email protected]")
.setAge("31")
.build();Пример класса UserAvroProducer:
package com.kafka.producer;
import com.kafka.utils.DefaultProperties;
import modelAvro.user.UserAvro;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
public class UserAvroProducer {
public static void sendTopicMessage(String topic, UserAvro message) {
String generatedKey = String.valueOf(Math.random());
Producer<String, UserAvro> producer = new KafkaProducer<>(DefaultProperties.getProducerAvroProperties());
try {
producer.send(new ProducerRecord<>(topic, generatedKey, message));
System.out.println("Avro-сообщение отправлено в топик: " + message.toString());
} finally {
producer.close();
}
}
}Пример класса UserAvroConsumer:
package com.kafka.consumer;
import com.kafka.utils.DefaultProperties;
import modelAvro.user.UserAvro;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
public class UserAvroConsumer {
public static List<String> getTopicMessages(String topic) {
List<String> messages = new ArrayList<>();
Consumer<String, UserAvro> consumer = new KafkaConsumer<>(DefaultProperties.getConsumerAvroProperties(topic));
consumer.subscribe(Collections.singletonList(topic));
try {
ConsumerRecords<String, UserAvro> records = consumer.poll(Duration.ofSeconds(10));
records.forEach(record -> {
String messageValue = record.value().toString();
messages.add(messageValue);
System.out.println("Получено Avro-сообщение из топика: " + messageValue);
});
consumer.commitAsync();
} catch (Exception e) {
System.err.println("Ошибка при потреблении сообщений из топика: " + e.getMessage());
} finally {
consumer.close();
}
return messages;
}
}В пакете com.kafka.tests находится класс KafkaTests, содержащий интеграционные тесты для проверки работы Producers и
Consumers.
Для запуска тестов выполните команду:
mvn clean testТесты проверяют отправку и получение сообщений как в стандартном формате JSON, так и в формате Avro.
Confluent Control Center предоставляет визуальный интерфейс для мониторинга и управления кластером Kafka, включая топики, схемы и сообщения.
В docker-compose.yml уже настроен сервис control-center. Для доступа откройте браузер и перейдите по
адресу http://localhost:9021.