0

我正在关注来自 github 的消息中心 java 示例。我创建了自己的 Kafka 生产者版本，它在使用 java 主应用程序的 J2Se 环境中运行良好。下面的代码片段 -

String jasslocation = "C:\\HOME\\Technical\\bluemix\\git\\message-hub-samples\\java\\message-hub-kafka-ssl\\resources" + File.separator + "jaas.conf";
		System.setProperty(JAAS_CONFIG_PROPERTY, jasslocation);
		
		Properties props = new Properties();
        props.put("key.serializer","org.apache.kafka.common.serialization.ByteArraySerializer");
        props.put("value.serializer","org.apache.kafka.common.serialization.ByteArraySerializer");
        props.put("bootstrap.servers","kafka02-prod02.messagehub.services.eu-gb.bluemix.net:9093");
        props.put("acks","all");
     //   props.put("block.on.bu","true");
        props.put("batch.size","1");
        props.put("security.protocol","SASL_SSL");
        props.put("ssl.protocol","TLSv1.2");
        props.put("ssl.enabled.protocols","TLSv1.2");
        props.put("ssl.truststore.location","C:\\IBM\\BPM857\\java_1.7_64\\jre\\lib\\security\\cacerts");
        //props.put("ssl.truststore.location","C:\\IBM\\IID\\v8.5\\jdk\\jre\\lib\\security\\cacerts");
        
        props.put("ssl.truststore.password","changeit");
        props.put("ssl.truststore.type","JKS");
        props.put("ssl.endpoint.identification.algorithm","HTTPS");
        
        String topic="mytopic";
        String message="publish message to a topic";
        try {
        	
          KafkaProducer<byte[], byte[]> kafkaProducer;
          kafkaProducer = new KafkaProducer<byte[], byte[]>(props);
          ProducerRecord<byte[], byte[]> producerRecord = new ProducerRecord<byte[], byte[]>(topic,message.getBytes("UTF-8"));
         
          RecordMetadata recordMetadata = kafkaProducer.send(producerRecord).get();
          //getting RecordMetadata is possible to validate topic, partition and offset
          System.out.println("topic where message is published : " + recordMetadata.topic());
          System.out.println("partition where message is published : " + recordMetadata.partition());
          System.out.println("message offset # : " + recordMetadata.offset());
          kafkaProducer.close();

kafka 客户端版本 - 0.90

当我将此代码包装在 J2EE MDB 中并将其部署在 BPM 服务器上时。当 MDB 触发时，线程继续等待 send() ：-

RecordMetadata recordMetadata = kafkaProducer.send(producerRecord).get();

并在一段时间后超时。它不会抛出任何其他异常或错误。

我的问题是，为了从 BPM 内部使用 kafka 客户端 api，我是否必须执行任何其他步骤，例如 ssl 配置等。

请指教。

谢谢。

4

0 回答 0