Tag: KafkaProducer

Kafka & Java: Secured Producer Send Record

In this tutorial I will show you how to put a record to a secured Kafka. Before you begin you will need Maven/Eclipse all setup and a project ready to go. If you haven’t installed Kafka Kerberos yet please do so.

Import SSL Cert to Java:

Follow this tutorial to “Installing unlimited strength encryption Java libraries”

If on Windows do the following

#Import it
"C:\Program Files\Java\jdk1.8.0_171\bin\keytool" -import -file hadoop.csr -keystore "C:\Program Files\Java\jdk1.8.0_171\jre\lib\security\cacerts" -alias "hadoop"

#Check it
"C:\Program Files\Java\jdk1.8.0_171\bin\keytool" -list -v -keystore "C:\Program Files\Java\jdk1.8.0_171\jre\lib\security\cacerts"

#If you want to delete it
"C:\Program Files\Java\jdk1.8.0_171\bin\keytool" -delete -alias hadoop -keystore "C:\Program Files\Java\jdk1.8.0_171\jre\lib\security\cacerts"

POM.xml

<dependency>
	<groupId>org.apache.kafka</groupId>
	<artifactId>kafka-clients</artifactId>
	<version>1.1.0</version>
</dependency>

Imports

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

Producer JAAS Conf (client_jaas.conf)

KafkaClient {
    com.sun.security.auth.module.Krb5LoginModule required
    useTicketCache=false
    refreshKrb5Config=true
    debug=true
    useKeyTab=true
    storeKey=true
    keyTab="c:\\data\\kafka.service.keytab"
    principal="kafka/hadoop@REALM.CA";
};

Producer Props File

You can go here to view all the options for producer properties.

bootstrap.servers=hadoop:9094
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer
security.protocol=SASL_SSL
sasl.kerberos.service.name=kafka

Initiate Kerberos Authentication

System.setProperty("java.security.auth.login.config", "C:\\data\\kafkaconnect\\kafka\\src\\main\\resources\\client_jaas.conf");
System.setProperty("https.protocols", "TLSv1,TLSv1.1,TLSv1.2");
System.setProperty("java.security.krb5.conf", "C:\\Program Files\\Java\\jdk1.8.0_171\\jre\\lib\\security\\krb5.conf");
System.setProperty("java.security.krb5.realm", "REALM.CA");
System.setProperty("java.security.krb5.kdc", "REALM.CA");
System.setProperty("sun.security.krb5.debug", "false");
System.setProperty("javax.net.debug", "false");
System.setProperty("javax.net.ssl.keyStorePassword", "changeit");
System.setProperty("javax.net.ssl.keyStore", "C:\\Program Files\\Java\\jdk1.8.0_171\\jre\\lib\\security\\cacerts");
System.setProperty("javax.net.ssl.trustStore", "C:\\Program Files\\Java\\jdk1.8.0_171\\jre\\lib\\security\\cacerts");
System.setProperty("javax.net.ssl.trustStorePassword", "changeit");
System.setProperty("javax.security.auth.useSubjectCredsOnly", "true");

Producer Connection/Send

The record we will send will just be a string for both key and value.

Producer<String, String> producer = null;

try {
	ClassLoader classLoader = getClass().getClassLoader();
	//Get the props file and load to the producer.
	try (InputStream props = classLoader.getResourceAsStream("producer.props")) {
		Properties properties = new Properties();
		properties.load(props);
		producer = new KafkaProducer<>(properties);
	}
	//Setting up the record to send
	ProducerRecord<String, String> rec = new ProducerRecord<String, String>("testTopic", "Key", "Value");
	//Send the record and get the response
	RecordMetadata recMetaData = producer.send(rec).get();

	//You can now print out any relavent information you want about the RecordMetaData

	System.out.println("Producer Record Sent");
} finally {
	producer.flush();
	producer.close();
}

References

I used kafka-sample-programs as a guide for setting up props.

Kafka & Java: Unsecure Producer Send Record

In this tutorial I will show you how to put a record to Kafka. Before you begin you will need Maven/Eclipse all setup and a project ready to go.  If you haven’t installed Kafka yet please do so.

POM.xml

<dependency>
	<groupId>org.apache.kafka</groupId>
	<artifactId>kafka-clients</artifactId>
	<version>1.1.0</version>
</dependency>

Imports

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

Producer Props File

You can go here to view all the options for producer properties.

# The url to kafka
bootstrap.servers=localhost:9092

# The number of acknowledgments the producer requires the leader to have received before considering a request complete
acks=all

# Setting a value greater than zero will cause the client to resend any record whose send fails with a potentially transient error
retries=1

# The producer will attempt to batch records together into fewer requests whenever multiple records are being sent to the same partition. 
batch.size=16384

# The frequency in milliseconds that the consumer offsets are auto-committed to Kafka 
auto.commit.interval.ms=1000

# The producer groups together any records that arrive in between request transmissions into a single batched request. Normally this occurs only under load when records arrive faster than they can be sent out
linger.ms=0

# The serializer for the key
key.serializer=org.apache.kafka.common.serialization.StringSerializer

# The serializer for the value
value.serializer=org.apache.kafka.common.serialization.StringSerializer


# The configuration controls the maximum amount of time the client will wait for the response of a request
request.timeout.ms=60000

# If true the consumer's offset will be periodically committed in the background.
enable.auto.commit=true

Producer Connection/Send

The record we will send will just be a string for both key and value.

Producer<String, String> producer = null;

try {
	ClassLoader classLoader = getClass().getClassLoader();
	//Get the props file and load to the producer.
	try (InputStream props = classLoader.getResourceAsStream("producer.props")) {
		Properties properties = new Properties();
		properties.load(props);
		producer = new KafkaProducer<>(properties);
	}
	//Setting up the record to send
	ProducerRecord<String, String> rec = new ProducerRecord<String, String>("testTopic", "Key", "Value");
	//Send the record and get the response
	RecordMetadata recMetaData = producer.send(rec).get();

	//You can now print out any relavent information you want about the RecordMetaData

	System.out.println("Producer Record Sent");
} finally {
	producer.flush();
	producer.close();
}

References

I used kafka-sample-programs as a guide for setting up props.