Сериализация данных с Apache Avro в Kafka

byte[] serialize(String topic, T data);

T deserialize(String topic, byte[] data);

public class JsonSerializer<T> implements Serializer<T> {
    private final ObjectMapper objectMapper = new ObjectMapper();
    
    @Override
    public byte[] serialize(String topic, T data) {
        try {
            return objectMapper.writeValueAsBytes(data);
        } catch (Exception e) {
            throw new SerializationException("Ошибка сериализации в JSON", e);
        }
    }
    
    // другие методы...
}

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

// Создаём схему из строки JSON
Schema schema = new Schema.Parser().parse(schemaJson);
 
// Создаём запись на основе схемы
GenericRecord user = new GenericData.Record(schema);
user.put("id", 123);
user.put("name", "Иван Петров");
user.put("email", "ivan@example.com");
 
// Сериализуем запись в байтовый массив
ByteArrayOutputStream out = new ByteArrayOutputStream();
BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(out, null);
GenericDatumWriter<GenericRecord> writer = new GenericDatumWriter<>(schema);
writer.write(user, encoder);
encoder.flush();
 
byte[] serializedData = out.toByteArray();

// Используем ту же схему для десериализации
ByteArrayInputStream in = new ByteArrayInputStream(serializedData);
BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(in, null);
GenericDatumReader<GenericRecord> reader = new GenericDatumReader<>(schema);
GenericRecord result = reader.read(null, decoder);
 
// Теперь можем получить доступ к полям
int id = (Integer) result.get("id");
String name = result.get("name").toString();

// Создаём объект сгенерированного класса
User user = new User();
user.setId(123);
user.setName("Иван Петров");
user.setEmail("ivan@example.com");
 
// Сериализуем объект
ByteArrayOutputStream out = new ByteArrayOutputStream();
BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(out, null);
SpecificDatumWriter<User> writer = new SpecificDatumWriter<>(User.class);
writer.write(user, encoder);
encoder.flush();

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
props.put("schema.registry.url", "http://localhost:8081");
 
KafkaProducer<String, GenericRecord> producer = new KafkaProducer<>(props);
 
// Создаем запись Avro
GenericRecord user = new GenericData.Record(schema);
user.put("id", 1);
user.put("name", "Иван Сидоров");
user.put("email", "ivan.sidorov@example.com");
 
// Отправляем сообщение
ProducerRecord<String, GenericRecord> record = 
    new ProducerRecord<>("users", user);
producer.send(record);
producer.close();

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "user-consumer-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
props.put("schema.registry.url", "http://localhost:8081");
props.put("specific.avro.reader", "true");
 
KafkaConsumer<String, User> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("users"));
 
while (true) {
    ConsumerRecords<String, User> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, User> record : records) {
        User user = record.value();
        System.out.println("Получен пользователь: " + user.getName());
    }
}

curl -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" \
--data '{"compatibility": "BACKWARD"}' \
http://localhost:8081/config

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
props.put("schema.registry.url", "http://localhost:8081");
 
KafkaProducer<String, User> producer = new KafkaProducer<>(props);
 
// Создаем объект User через сгенерированный класс
User user = User.newBuilder()
    .setId(42)
    .setName("Алексей Смирнов")
    .setEmail("alex@example.com")
    .build();
 
// Отправляем сообщение
ProducerRecord<String, User> record = new ProducerRecord<>("users", user);
producer.send(record);

props.put("schema.registry.url", "https://schema-registry:8081");
props.put("schema.registry.ssl.truststore.location", "/path/to/truststore.jks");
props.put("schema.registry.ssl.truststore.password", "[B][/B][B][/B]**");

props.put("basic.auth.credentials.source", "USER_INFO");
props.put("schema.registry.basic.auth.user.info", "username:password");

// Пример обработки события изменения статуса заказа
void processOrderStatusChange(byte[] messageBytes) {
    try {
        // Десериализуем сообщение с использованием схемы из реестра
        OrderStatusChangedEvent event = 
            avroDeserializer.deserialize("order-status-topic", messageBytes);
            
        // Обработка события в зависимости от нового статуса
        switch (event.getNewStatus().toString()) {
            case "PAID":
                initiateShipping(event.getOrderId());
                break;
            case "SHIPPED":
                updateCustomerNotification(event.getOrderId(), "Ваш заказ отправлен!");
                break;
            // другие статусы...
        }
    } catch (AvroSchemaParseException e) {
        // Обработка ошибок схемы
        logSchemaError(e);
    } catch (Exception e) {
        // Другие ошибки
        logGenericError(e);
    }
}