Kafka JMS How-to using Confluent JMS client

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

  1. Start Kafka Broker

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

  1. Create Topic

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

  1. Download confluent JMS client library with all it’s dependencies, for the demo the following libraries are used
  • jose4j-0.7.9.jar
  • kafka-jms-client-7.0.1.jar
  • gf.javax.jms.jar
  • connect-runtime-2.7.2.jar
  • protobuf-java.jar
  • protobuf-java-util.jar
  • connect-api-3.0.0.jar
  • kafka-clients-2.7.2.jar
  • com.fasterxml.jackson.databind.jar
  • jackson-annotations-2.1.2.jar
  • jackson-core-2.9.9.jar
  • guava-31.0-jre.jar
  1. JNDI configuration

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

  1. We could either use a JNDI properties file (with the below contents and provide the file location in Provider URL section) or directly provide them into the JNDI context (see the sample program below), like
    java.naming.factory.initial=io.confluent.kafka.jms.KafkaInitialContextFactory
    bootstrap.servers=localhost:9092
    topic.t1=t1 #topic destination
    queue.q1=q1 #queue destination
    confluent.topic.replication.factor=3 (change based on requirement)
    client.id=my-test-client #JMS connection client ID

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.

Demo

kafkajms

2 Likes

Good one @Prasanta_Malik might help many users who are looking to leverage the capabilities of IS and Kafka.

Thanks,
Sree