import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.config.SaslConfigs;

import java.io.*;
import java.sql.Timestamp;
import java.util.Date;
import java.util.Properties;
import java.util.logging.Level;
import java.util.logging.LogManager;
import java.util.logging.Logger;

public class KafkaProducerCustom {
    private static Properties props;
    private static KafkaProducer kafkaProducer;
    private final static String propFileName = "/u01/app/oracle/kafka/kafka.properties";
    private final static StringWriter errors = new StringWriter();


    static {
        try {

            FileInputStream fis = new FileInputStream(propFileName);
            if (fis != null) {
                props = new Properties();
                props.load(fis);
                props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
                props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
				// остальные строки получения значений из файла конфига убрал
                kafkaProducer = new KafkaProducer(props);
                fis.close();
            } else {
                props = null;
                kafkaProducer = null;
            }
        } catch (Exception e) {
            e.printStackTrace(new PrintWriter(errors));
        }
    }

    public static String sendMessage(Object topicName, Object value) {

        final StringWriter errors = new StringWriter();

        Integer partition = props.getProperty("partition") == null ? null : Integer.valueOf(props.getProperty("partition"));

        ProducerRecord<Object, Object> record = partition == null ?
                new ProducerRecord<Object, Object>((String) topicName, value) :
                new ProducerRecord<Object, Object>((String) topicName, partition, value);

        try {
            kafkaProducer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (enablelogger){
                        LOGGER.info("onCompletion() inside");
                    }
                    if (enablelogger && recordMetadata != null){
                        LOGGER.info("Logging the response");
                        LOGGER.info("Successfully received the details as: " +
                                "   Topic: " + recordMetadata.topic() +
                                "   Partition: " + recordMetadata.partition() +
                                "   Offset: " + recordMetadata.offset() +
                                "   Timestamp: " + new Date(new Timestamp(recordMetadata.timestamp()).getTime()));
                    }
                }
            }).get();

    
        } catch (Exception e) {
            e.printStackTrace(new PrintWriter(errors));
            return errors.toString();
        }
        return "";
    }

    public static void main(String[] args) {
        System.out.println(KafkaProducerCustom.sendMessage("", ""));
    }
}