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 removingC:\tmp\kafka-logsdirectory, 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
-
Download Apache Kafka
-
Extract to a location (ex: C:\kafka\kafka_2.12-2.8.1)
-
Start ZooKeeper service
C:\kafka\kafka_2.12-2.8.1\bin\windows>zookeeper-server-start.bat …/…/config/zookeeper.properties
- Start Kafka Broker
C:\kafka\kafka_2.12-2.8.1\bin\windows>kafka-server-start.bat …/…/config/server.properties
- Create Topic
C:\kafka\kafka_2.12-2.8.1\bin\windows>kafka-topics.bat --create --topic t1 --bootstrap-server localhost:9092
- 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
- JNDI configuration
Initial context factory: io.confluent.kafka.jms.KafkaInitialContextFactory
Consider providing all other properties like Topic, client id etc…
- 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
-
For JNDI configuration, provide all properties in “Other Properties” section
-
While creating the JMS connection, uncheck the “Create Temporary Queue” option as Kafka does not have a concept of a temporary queue.
-
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
