PaymentConfirmedProducer

PaymentConfirmedProducer.java

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.security.scram.ScramLoginModule;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class PaymentConfirmedProducer {

    private static final String _paymentConfirmedTopic = "payment-confirmed";

    public static void main(String[] args) {
        try (KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(_getProperties())) {
            kafkaProducer.send(
                    new ProducerRecord<>(_paymentConfirmedTopic, "{\"orderERC\":\"foo\"}"));

            kafkaProducer.flush();
        }
    }

    private static Properties _getProperties() {
        Properties properties = new Properties();

        properties.put("sasl.jaas.config", ScramLoginModule.class.getName() + " required username=\"" + _userName + "\" password=\"" +  _password + "\";");
        properties.put("sasl.mechanism", "SCRAM-SHA-256");
        properties.put("security.protocol", "SASL_SSL");

        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, _bootstrapServer);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        return properties;
    }

    private final static String _bootstrapServer = "...";
    private final static String _password = "...";
    private final static String _userName = "...";
}