Tagged KAFKA

Building Producer and Consumer Clients in go Language for Oracle Event Hub Cloud Service


CREATE AN OEH CLUSTER

Documentation here or watch following video:

When the cluster is created, go to details page and grab the connection url, which has the  format <broker1_ip>:6667,…,<brokern_ip>:6667

CREATE A TOPIC

INSTALL THE LIB

I’ve done the following (Mac):

git clone https://github.com/edenhill/librdkafka.git
cd librdkafka
./configure --prefix /usr/local
make
sudo make install (sudo make uninstall in case you want to remove the lib)

brew install librdkafka

BUILD OR GRAB A SAMPLE CLIENT

I’ve downloaded this consumer and this producer samples. Create producer.go and consumer.go files and put the sample code in each file.

go build producer.go

go build consumer.go

Test your clients:

./producer <broker1_ip>:6667,...,<brokern_ip>:6667 sample

./consumer <broker1_ip>:6667,...,<brokern_ip>:6667 consumer_example sample

kgp.png

goc

That’s all folks!

🙂 Enjoy

 

jMeter Test Script for Kafka (tested against Oracle Event Hub Cloud Service)


In this post we are testing a jMeter script that sends messages to Kafka Cluster running in Oracle Event Hub (OEH).

jMETER CONFIG

Install plugins manager in JMeter if you don’t have it, restart jMeter and then install Kafka Support, then you will be told to restart jMeter.

a1

jMETER SCRIPT

Download the jMeter script from here

Load the script into jMeter and change the configuration of your cluster, topic name, etc..

a3

Change the “View Results Tree” configuration ( it comes with theErrors Check enabled in order to avoid latencies in test execution, because the nature of this especific listener ). Execute a few test cycles and observe that valid son files are sent to Kafka cluster.

a2.png

Once you are ready, put the checkbox in Errors and execute the script, here you can watch a video with my test. It has been rum from my local network to cloud, therefore latencies are degrading performance.

 

That’s all folks!

Enjoy 😉

Creating a Fast&Simple Container for Sending Messages to a Topic in Oracle Event Hub Cloud Service (aka OEHCS, which is a Kafka cluster) and Deploying it to Kubernetes Cluster


The container uses 4 environment variables, you can find a container already built for you here

SOURCE CODE OF THE PRODUCER

var sleep = require('system-sleep');

const oehcs_connect_url = process.env.OEHCS_CONNECTURL

const topic_name = process.env.TOPIC_NAME

const num_partitions = process.env.NUM_PARTITIONS

const message = process.env.MESSAGE

var kafka = require('kafka-node'),

HighLevelProducer = kafka.HighLevelProducer,

client = new kafka.KafkaClient({kafkaHost: oehcs_connect_url}),

producer = new HighLevelProducer(client);

var i = 0;

while (i >= 0 ){

var payloads = [{ topic: topic_name, messages: message , partition: i}];

//producer.on('ready', function () {

producer.send(payloads, function (err, data) {

console.log(data);

});

// });

i = i + 1;

if (i > num_partitions -1){

i = 0;

sleep( 1 );

}

}

THE DOCKERFILE

FROM oraclelinux:7-slim
WORKDIR /app
ADD . /app
RUN curl --silent --location https://rpm.nodesource.com/setup_10.x | bash -
RUN yum -y install nodejs npm
CMD ["node","producer-direct.js"]

THE YAML FOR K8S DEPLOYMENT

apiVersion: apps/v1

kind: Deployment

metadata:

name: oehcsnodeproducer-direct

labels:

app: oehcsnodeproducer-direct

spec:

replicas: 1

selector:

matchLabels:

app: oehcsnodeproducer-direct

strategy:

type: Recreate

template:

metadata:

labels:

app: oehcsnodeproducer-direct

spec:

containers:

- image: javiermugueta/oehcsnodeproducer-direct

env:

- name: OEHCS_CONNECTURL

value: "<ip1>:6667,<ip2>:6667,..."

- name: TOPIC_NAME

value: "R1"

- name: NUM_PARTITIONS

value: "10"

- name: MESSAGE

value: "{'put here what you want'}"

name: oehcsnodeproducer-direct

TEST IT AND SEE WHAT HAPPENS

Create the deployment and after 10 minutes take a look to production messages ratio:
kubectl apply -f my.yaml
oehcs-scale1.png
More or less 400/second…
Scale the deployment and take a look to new production ratios:
kubectl scale deployment oehcsnodeproducer-direct --replicas=2
Around 8000 messages/second!
Now add 9 partitions to the topic and take a look to new ratios:
add partitions
With 2 pods running and 10 partitions we are producing around 10K messages per second! As you can see partitioning improves performance!
10partitions.png
Let’s double the number of pods and see new ratios:
kubectl scale deployment oehcsnodeproducer-direct --replicas=4
And now 18K/second!
18k
That’s all folks!
Enjoy 😉

Mirroring a Topic Between 2 Oracle Event Hub (Kafka) Clusters in 20 Minutes


Oracle Event Hub Cloud Service (OEHCS) is a Kafka managed PaaS cloud service. In a few minutes you can provision a full Kafka cluster ready for creating topics for sending and consuming messages.

mirror

In this post we will configure a feature called ‘mirroring’ that allows you to replicate the messages that are sent to a topic in a source cluster to another topic in a destination cluster.

FIRST

Create 2 OEHCS instances, the names and sizes you want. See this post for help (coming soon).

oehcs1

Create a topic in the origin cluster, name it R1.

oehcs2

Create a topic in the destination cluster, name it R1R1.

oehcs3

SECOND

Grab the public IP of one of the brokers in the destination cluster. ssh to that server and follow what is stated here (except for the point number 10)

oehcs5

Configuration files examples:

mkdir -p /u01/oehpcs/confluent/etc/mirror-maker

vi sourceClusterConsumer.config

bootstrap.servers=130.61.36.87:6667,130.61.86.29:6667
group.id=replica-consumer
exclude.internal.topics=true
auto.offset.reset=earliest
partition.assignment.strategy=org.apache.kafka.clients.consumer.RoundRobinAssignor

vi targetClusterProducer.config

bootstrap.servers=130.61.46.183:6667,130.61.80.227:6667
acks=-1
max.in.flight.requests.per.connection=1
compression.type=none

Message for hackers: The IP’s were real one day but they don’t exist any more 😉

THIRD

Execute this command:

/u01/oehpcs/confluent/bin/kafka-mirror-maker.sh --consumer.config /u01/oehpcs/confluent/etc/mirror-maker/sourceClusterConsumer.config --producer.config /u01/oehpcs/confluent/etc/mirror-maker/targetClusterProducer.config --num.streams 2 --whitelist "R.*" --message.handler kafka.tools.OehcsTopicSuffixMirrorMakerHandler --message.handler.args R1

FOURTH

Produce messages to R1 topic in origin cluster the way you can. See this post for help.

oehcs6

Please note that a yellow band appears in the Throughput graph indicating that bytes are  going out the topic, who is extracting messages?

FIFTH

Notice that the R1R1 topic in destination cluster is receiving messages.

oehcs7

So far so good, we have replicated the topic in another cluster thousand miles away…

Stop the mirroring process, add partitions to R1R1 topic up to 20 and start the mirroring again with this new command:

/u01/oehpcs/confluent/bin/kafka-mirror-maker.sh --consumer.config /u01/oehpcs/confluent/etc/mirror-maker/sourceClusterConsumer.config --producer.config /u01/oehpcs/confluent/etc/mirror-maker/targetClusterProducer.config --num.streams 20 --whitelist "R.*" --message.handler kafka.tools.OehcsTopicSuffixMirrorMakerHandler --message.handler.args R1

mirrorscaled.png

As you can see now the mirroring runs faster!

That’s all folks!

Enjoy 😉