Kafka JMS How-to using Confluent JMS client
Tags: webmethods, integrationserver, esb
Published: 2022-03-08 10:59:49

---
title: Kafka JMS How-to using Confluent JMS client
published: true
date: 2022-02-21 14:26:46 UTC
tags: webMethods, IntegrationServer, ESB
canonical_url: https://tech.forums.softwareag.com/t/kafka-jms-how-to-using-confluent-jms-client/255391
---
>Note:
If you are trying out in windows, download the Kafka version kafka_2.12-2.8.1 (For other versions there are some permission issues in windows like accessing/writing to the tmp location where the kafka logs are stored). Try removing C:\tmp\kafka-logs directory, sometimes that helps to resolve the issue.

>Note:
SoftwareAG has not certified this combination officially.

### Follow the steps below to setup a producer/consumer test scenario

1. Download Apache Kafka

2. Extract to a location (ex: C:\kafka\kafka\_2.12-2.8.1)

3. Start ZooKeeper service

> C:\kafka\kafka\_2.12-2.8.1\bin\windows>zookeeper-server-start.bat …/…/config/zookeeper.properties

4. Start Kafka Broker

> C:\kafka\kafka\_2.12-2.8.1\bin\windows>kafka-server-start.bat …/…/config/server.properties

5. Create Topic

> C:\kafka\kafka\_2.12-2.8.1\bin\windows>kafka-topics.bat --create --topic t1 --bootstrap-server localhost:9092

6. Download confluent JMS client library with all it’s dependencies, mostly the following list

- guava
- connect api
- connect runtime
- JMS spec 1.1
- jose4j
- kafka clients
- kafka jms client
- protobuf
- protobuf java util
- jackson

7. JNDI configuration  
Initial context factory: io.confluent.kafka.jms.KafkaInitialContextFactory  
Consider providing all other properties like Topic, client id etc…

8. We could either use a jndi properties file or directly provide them into the JNDI content properties, like  
java.naming.factory.initial=io.confluent.kafka.jms.KafkaInitialContextFactory  
bootstrap.servers=localhost:9092  
topic.t1=t1  
queue.q1=q1  
confluent.topic.replication.factor=3 (change based on requirement)  
client.id=my-test-client

### Sample program to test the connection

```
import java.util.Properties;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

public class KafkaConnect {

	private static final String BOOTSTRAP_SERVERS = "localhost:9092";
	private static final String TOPIC = "t1";
	private static final String CLIENT_ID = "testClient";

	public static void main(String[] args) {

		Connection conn = null;
		InitialContext _jndiContext = null;

		try {
			Properties props = new Properties();

			props.put("java.naming.factory.initial", "io.confluent.kafka.jms.KafkaInitialContextFactory");

			props.put("bootstrap.servers", "localhost:9092");
			props.put("topic.t1", "t1");
			props.put("queue.q1", "q1");
			props.put("confluent.topic.replication.factor", "3");
			props.put("client.id", "my-test-client");

			// Initialize the JNDI context
			Context ctx = new InitialContext(props);
			System.out.println("-----Initial Content Initialized------");

			// Lookup the connection factory to get a connection
			_jndiContext = new InitialContext(props);
			ConnectionFactory cf = (ConnectionFactory) _jndiContext.lookup("ConnectionFactory");
			System.out.println("-----Connection Factory lookup success-----");

			// Start the JMS connection from the factory
			conn = cf.createConnection();
			conn.start();
			System.out.println("-----Connection Started-----" + conn);
			
//This is just a sample, change the producer and consumer code according to your need
			new ProduceAndConsumerMessage(conn, (Destination) _jndiContext.lookup("q1")).sendReceive();

			conn.close();
			System.out.println("-----Connection Closed-----");
		} catch (Exception ex) {
			ex.printStackTrace();
		} finally {
			if (_jndiContext != null) {
				try {
					_jndiContext.close();
				} catch (NamingException e) {
					e.printStackTrace();
				}
			}
			if (conn != null) {
				try {
					conn.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
		}

	}
}

class ProduceAndConsumerMessage {

	Connection connection;
	Destination destination;
	javax.jms.MessageProducer producer;
	Session session;
	javax.jms.MessageConsumer consumer;
	
	public ProduceAndConsumerMessage(Connection connection, Destination destination) throws JMSException {

		this.connection = connection;
		this.destination = destination;		
	}

	public void sendReceive() throws JMSException {

		try {
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			
			TextMessage message = session.createTextMessage();
			producer = session.createProducer(destination);
			producer.send(destination, message);
			System.out.println("Sent message");

			//consume
			consumer = session.createConsumer(destination);
			Message msg = consumer.receive(500);
			if (msg != null) {
				System.out.println("Received message: " + msg.getJMSMessageID());
			}
		} finally {
			if (producer != null) {
				producer.close();
			}
			if (consumer != null) {
				consumer.close();
			}
			if (session != null) {
				session.close();
			}			
		}
	}
}

```

### Trying out with Integration Server JMS

1. For JNDI configuration, provide all properties in “Other Properties” section  

2. While creating the JMS connection, uncheck the “Create Temporary Queue” option as Kafka does not have a concept of a temporary queue.

3. Use pub.jms:send to send a message to a destination


> Note: 
Currently the Integration Server JMS consumer implementation uses the javax.jms.Session.createConsumer(Destination, String) JMS API to create a consumer with a message selector, which kafka doesn’t support. So you can’t use JMS triggers for this combination and to add this support contact the product team.

Visit the [original article](https://tech.forums.softwareag.com/t/kafka-jms-how-to-using-confluent-jms-client/255391) in the Software AG Tech Community to check out the **demo**.