import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.util.*;

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("enable.auto.commit", "false"); // Manual commit

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("topic"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // Commit offsets before losing partitions
        commitOffsets(); // Custom method to commit processed offsets
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // Restore state or seek to offsets
        /* warm-up: e.g., load local state for partitions */
    }
});

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        // Process records
        consumer.commitAsync(); // Or commitSync for sync
    }
} finally {
    consumer.close();
}

import org.apache.kafka.clients.producer.*;
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import java.util.Properties;

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092"); // Адреса серверов Kafka
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // Сериализатор для ключей
props.put("value.serializer", KafkaAvroSerializer.class.getName()); // Сериализатор для значений Avro
props.put("schema.registry.url", "http://schema-registry:8081"); // URL реестра схем

KafkaProducer<String, GenericRecord> producer = new KafkaProducer<>(props);

// Пример схемы и записи
String schemaStr = "{\"type\":\"record\",\"name\":\"Order\",\"fields\":[{\"name\":\"id\",\"type\":\"int\"},{\"name\":\"amount\",\"type\":\"double\"}]}";
Schema schema = new Schema.Parser().parse(schemaStr); // Разбор схемы
GenericRecord record = new GenericData.Record(schema); // Создание записи
record.put("id", 1); // Заполнение поля id
record.put("amount", 100.5); // Заполнение поля amount

ProducerRecord<String, GenericRecord> producerRecord = new ProducerRecord<>("orders", "key", record); // Создание записи для отправки
producer.send(producerRecord); // Отправка
producer.close(); // Закрытие отправителя