Tag: producer

Azure: EventHub

In this tutorial I will show you how to connect to event hub from Python. Ensure you have first installed an IDE (Eclipse) and Python3.7.

Python Package Installation

pip3 install azure-eventhub

Create a Producer

This will publish events to event hub. The important part here is the “EndPoint”. You need to login to Azure Portal and get the get the endpoint from the “Shared Access Policies” from the event hub namespace.

from azure.eventhub import EventHubProducerClient, EventData, EventHubConsumerClient

connection_str = 'Endpoint=sb://testeventhubnamespace.servicebus.windows.net/;SharedAccessKeyName=<<THE_ACCESS_KEY_NAME>>;SharedAccessKey=<<THE_ACCESS_KEY>>'
eventhub_name = '<<THE_EVENT_HUB_NAME>>'
producer = EventHubProducerClient.from_connection_string(connection_str, eventhub_name=eventhub_name)

event_data_batch = producer.create_batch()

event_data_batch.add(EventData('My Test Data'))

with producer:
    producer.send_batch(event_data_batch)

Create a Consumer

This will monitor the event hub for new messages.

from azure.eventhub import EventHubProducerClient, EventData, EventHubConsumerClient

connection_str = 'Endpoint=sb://testeventhubnamespace.servicebus.windows.net/;SharedAccessKeyName=<<THE_ACCESS_KEY_NAME>>;SharedAccessKey=<<THE_ACCESS_KEY>>'
eventhub_name = '<<THE_EVENT_HUB_NAME>>'
consumer_group = '<<THE_EVENT_HUB_CONSUMER_GROUP>>'
client = EventHubConsumerClient.from_connection_string(connection_str, consumer_group, eventhub_name=eventhub_name)

def on_event(partition_context, event):
    print("Received event from partition {} - {}".format(partition_context.partition_id, event))
    partition_context.update_checkpoint(event)

with client:
    #client.receive(
    #    on_event=on_event, 
    #    starting_position="-1",  # "-1" is from the beginning of the partition.
    #)
    client.receive(
        on_event=on_event
    )

 

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.