By any chance is anyone here using kafka with cf-2021? At one point I had a working example in cf-2016 or cf-2018, but I am not finding my old proof-of-concept code and I cannot remember the magic used to get it working. Sigh. https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients

@Steve I don't know if you're still just experimenting, or already bet the farm. I can't say I've used Kafka yet so hopefully someone can speak to that, but if you are just starting to look at queues, I highly recommend RabbitMQ. I wrote the CFML RabbitSDK for it and really love it plus the websocket add-ons via STOMP.

Its like anything, wrap it and slap it. <cfset props = createObject("java", "java.util.Properties")> <cfset props.put("bootstrap.servers", "localhost:9092")> <cfset props.put("acks", "all")> <cfset props.put("retries", 0)> <cfset props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")> <cfset props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")> <cfset producer = createObject("java", "org.apache.kafka.clients.producer.KafkaProducer").init(props)>

That pretty much mirrors my code, but I am getting various null or StringSerializer is not serializer exceptions which I believe are jar version mismatches.

first dump what ever you are trying to do, into a command line format. This helps when trying to figure out if its coldfusion not setting something, or sending something wrapped in the wrong character. Usually its something adding an extra space or something equally as weird.

As for RabbitMQ, that is an option. I prefer the consumer "pull" model that kafka uses over the "push" model that RMQ uses. I have A LOT or experience with the push method as we wrote our own before MQ's were a thing. In a not-so-perfect environment, push can send a message to a client that has one or more dead worker threads so client receives, but the request goes to a dead thread. In my mind, flipping this so the worker thread does the pull better guarantees the thread is not dead - it would not have pulled (polled).

You would use this as a basis for that, if you havent already (This is mega generic and still not 100 percent) <cfset props = createObject("java", "java.util.Properties")> <cfset props.put("bootstrap.servers", "localhost:9092")> <cfset props.put("acks", "all")> <cfset props.put("retries", 0)> <cfset props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")> <cfset props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")> <cfset producer = createObject("java", "org.apache.kafka.clients.producer.KafkaProducer").init(props)> <!--- start the producer thread ---> <cfset producerThread = createObject("java", "java.lang.Thread").init(function() { try { while (true) { <!--- do Kafka producer operations here, such as send() or poll() ---> } } catch (any e) { if (e.getClass().getName() == "java.lang.InterruptedException") { <!--- handle the interrupt, such as gracefully shutting down the producer ---> } } })> <cfset producerThread.start()> <!--- send the kill thread command ---> <cfset producerThread.interrupt()> <cfset producer.wakeup()>

Listening to messages with Rabbit is as simple as

var channel = rabbitClient
		.startConsumer( 
			queue='myQueue',
			autoAcknowledge=false,
			consumer=(message, channel, log)=>{
				<http://log.info|log.info>( 'Consumer 1 Message received: #message.getBody()#' );
				message.acknowledge();
			},
			error=(message, channel, log, exception)=>{
				log.error( 'Error processing message #message.getBody()#.  Error message: #exception.message#' );
			} );

That spins off a thread that sits and processes new messages as they come in.

RE: Sample producer. After a couple comment syntax corrections, I get the follwoing: Invalid value org.apache.kafka.common.serialization.StringSerializer for configuration key.serializer: Class org.apache.kafka.common.serialization.StringSerializer could not be found.

After installing any third-party jars, you should always restart the ColdFusion engine instance as well as its container server (IE restart tomcat in 99.99 percent of all cases). Though its more than likely a config issue with how you are initing Kafka.

Thread.currentThread().setContextClassLoader(null);
Producer<String, String> producer = new KafkaProducer(props);

This should fix the problem

I did attempt the setContextClassLoader route, that caused yet another issue for me, and several restarts in most all my tests. No joy. On the RabbitMQ path, I was able to get CF talking to RabbitMQ, fairly painlessly. I think the key difference is the java client drivers between the two, Kafka and RabbitMQ. The Kafka driver requires several jar dependencies to run whereas the RabbitMQ driver is self-sufficient, requiring no other jars. Now if I can find better RabbitMQ driver doc, the universe will be in balance. (I admit, I am a java novice, so reverse engineering CF-->java can sometimes be challenging for me so I need examples, not many examples here)

@Steve The only docs for the rabbitsdk are in the readme of the repo, which covers most all the use cases. There is also a large set of unit tests you can look at. Or you can ask questions over in #box-products and I'll be happy to offer advice

My current confusion resolves around the Go samples that use request headers and body, and the Java examples that do not use the headers layer. I have CF working without headers. but I would like to take advantage of the headers but have been unsuccessful creating request properties - or even finding the properties class that CF-->Java maps.

@Steve Do you have a specific header you need to send?

I'll take a look at the SDK code. Earlier code I could not use directly as I am not using box and it had some other dependencies. The specific headers I'm interested in at the moment are content-type, reply-to and possibly adding my own for route tracking in the response. These are RPC calls.