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

import java.io.*;

class Data200825 implements Serializable {
    private static final long serialVersionUID = 1L;
    transient int x = 10;
    int y = 20;
}

public class Task200825 {
    public static void main(String[] args) throws Exception {
        Data200825 original = new Data200825();
        
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        new ObjectOutputStream(baos).writeObject(original);
        
        Data200825 restored = (Data200825) new ObjectInputStream(
            new ByteArrayInputStream(baos.toByteArray())).readObject();
        
        System.out.println(restored.x + " " + restored.y);
    }
}

class Wallet210825 {
    public int money;
    
    public Wallet210825(int money) {
        this.money = money;
    }
}

public class Task210825 {
    public static void main(String[] args) {
        Wallet210825 wallet = new Wallet210825(100);
        addMoney(wallet, 50);
        System.out.println(wallet.money);
    }
    
    public static void addMoney(Wallet210825 w, int amount) {
        w = new Wallet210825(w.money + amount);
    }
}

# Прослушиватели
listeners=SASL_SSL://:9092
advertised.listeners=SASL_SSL://broker-host:9092

# SSL настройки
ssl.keystore.location=/etc/kafka/secrets/kafka.keystore.jks
ssl.keystore.password=keystore-pass
ssl.truststore.location=/etc/kafka/secrets/kafka.truststore.jks
ssl.truststore.password=truststore-pass
ssl.client.auth=required  # Требовать клиентские сертификаты

# SASL
sasl.enabled.mechanisms=SCRAM-SHA-512
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512

# Для KRaft добавьте:
process.roles=broker,controller
controller.listener.names=CONTROLLER
listener.security.protocol.map=CONTROLLER:PLAINTEXT,SASL_SSL:SASL_SSL

# ACL
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
super.users=User:admin
allow.everyone.if.no.acl.found=false