import org.apache.kafka.clients.producer.*;
import java.util.Properties;

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");
props.put("acks", "all");
props.put("enable.idempotence", "true");
props.put("transactional.id", "order-tx-1");

try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
    producer.initTransactions();
    producer.beginTransaction();
    ProducerRecord<String, String> record = new ProducerRecord<>("orders", "key", "value");
    producer.send(record);
    producer.commitTransaction();
} catch (ProducerFencedException e) {
    // Handle fencing, e.g., abort and reinitialize
} catch (KafkaException e) {
    // Abort on error
    producer.abortTransaction();
}

public class Person {
    // Поля (состояние)
    private String name;  // Имя человека
    private int age;      // Возраст человека
}

Здесь name и age — поля, хранящие состояние объекта.
Они private, чтобы внешний код не мог напрямую изменять их (инкапсуляция).

public class Person {
    private String name;
    private int age;

    // Конструктор (инициализирует состояние)
    public Person(String name, int age) {
        this.name = name;  // this — ссылка на текущий объект
        this.age = age;
    }

    // Метод (поведение): Вывод информации
    public void introduce() {
        System.out.println("Привет, меня зовут " + name + ", мне " + age + " лет.");
    }

    // Геттер для возраста
    public int getAge() {
        return age;
    }

    // Сеттер для возраста (с проверкой)
    public void setAge(int age) {
        if (age > 0) {
            this.age = age;
        } else {
            System.out.println("Возраст не может быть отрицательным!");
        }
    }
}

Конструктор: Person(String name, int age) устанавливает начальное состояние.
Метод introduce(): Выводит информацию (поведение).
Геттер getAge(): Возвращает значение поля.
Сеттер setAge(int age): Изменяет поле с проверкой (инкапсуляция).

public class Main {
    public static void main(String[] args) {
        // Создание объекта
        Person person1 = new Person("Алексей", 35);

        // Вызов метода
        person1.introduce();  // Вывод: Привет, меня зовут Алексей, мне 35 лет.

        // Использование геттера и сеттера
        System.out.println("Текущий возраст: " + person1.getAge());
        person1.setAge(36);
        System.out.println("Новый возраст: " + person1.getAge());
    }
}

new Person("Алексей", 35): Создает объект и вызывает конструктор.
person1.introduce(): Вызывает метод объекта.

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();
}

public Person() {
    // Пусто, поля инициализируются значениями по умолчанию (null для объектов, 0 для чисел)
}

public class Person {
    private String name;
    private int age;

    // Конструктор с параметрами
    public Person(String name, int age) {
        this.name = name;
        this.age = age;
    }
}

При создании объекта new Person("Алексей", 35) конструктор присваивает значения полям.

public class Person {
    private String name;
    private int age;

    public Person(String name, int age) {
        this.name = name;  // this.name — поле класса, name — параметр
        this.age = age;
    }
}

Без this Java бы подумал, что name = name; — это присваивание параметра самому себе, что бесполезно.

public class Person {
    private String name;
    private int age;

    // Конструктор по умолчанию (без параметров)
    public Person() {
        this.name = "Неизвестный";
        this.age = 0;
    }

    // Конструктор с одним параметром
    public Person(String name) {
        this.name = name;
        this.age = 0;  // Возраст по умолчанию
    }

    // Конструктор с двумя параметрами
    public Person(String name, int age) {
        this.name = name;
        this.age = age;
    }

    // Вызов другого конструктора с помощью this()
    public Person(int age) {
        this("Неизвестный", age);  // Вызывает конструктор с двумя параметрами
    }
}

Теперь объекты можно создавать по-разному:Person p1 = new Person();               // Неизвестный, 0
Person p2 = new Person("Алексей");      // Алексей, 0
Person p3 = new Person("Алексей", 35);  // Алексей, 35
Person p4 = new Person(35);             // Неизвестный, 35

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(); // Закрытие отправителя