Hi,
I have a simple Apigee proxy with Java Callout which contains a very simple class to publish messages to Kafka Topic (I am trying to do this for poc purpose):
package com.aneesh.test;
import java.util.Properties;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import com.apigee.flow.execution.ExecutionContext;
import com.apigee.flow.execution.ExecutionResult;
import com.apigee.flow.message.MessageContext;
public class KafkaTest implements Execution{
public ExecutionResult execute(MessageContext messageContext, ExecutionContext executionContext) {
Producer<String, String> producer = null;
Properties props = null;
String topicName = “Kafka.Test.Topic”;
try {
props = new Properties();
props.put(“bootstrap.servers”, “:”);
props.put(“acks”, “all”);
props.put(“retries”, 0);
props.put(“batch.size”, 16384);
props.put(“linger.ms”, 1);
props.put(“buffer.memory”, 33554432);
props.put(“key.serializer”, “org.apache.kafka.common.serialization.StringSerializer”);
props.put(“value.serializer”, “org.apache.kafka.common.serialization.StringSerializer”);
producer = new KafkaProducer<String, String>(props);
String inputMessage = messageContext.getMessage().getContent();
if (inputMessage != null && inputMessage.length() > 0) {
producer.send(new ProducerRecord<String, String>(topicName, null, inputMessage));
} else {
messageContext.getMessage().setContent(“Input message is empty!!”);
}
return ExecutionResult.SUCCESS;
} catch (Exception e) {
messageContext.getMessage().setContent(“Error in java class!!” + e.getClass().getName() + ": " + e.getMessage());
return ExecutionResult.ABORT;
} finally {
if (producer != null) {
try {
producer.close();
} catch (Exception e) {
messageContext.getMessage().setContent(“Error in java class!!” + e.getClass().getName() + ": " + e.getMessage());
return ExecutionResult.ABORT;
}
}
}
}
}
Java Callout policy configuration:
<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
JC_Kafka_Conn
com.aneesh.test.KafkaTest
java://KafkaTest.jar
I uploaded the my custom jar and the dependent library “kafka-clients-1.0.0.jar” as below.
When i try to deploy the proxy to Apigee Edge dev region (cloud), its throwing me the following error:
Error in deployment for environment dev. The revision is deployed, but traffic cannot flow. com.apigee.kernel.exceptions.spi.UncheckedException{ code = application.bootstrap.FailedToConfigure, message = Configuration failed, associated contexts = []}
Can anybody please help me resolve?
Thanks,
Aneesh.