一次初始化 KafkaProducer 对象并由 Jmeter 脚本中的所有采样器使用相同的对象

Question

我正在尝试实现如下所述的场景： 我必须创建一个负载测试脚本我的异步 API 正在发布和使用来自 Kafka 主题的事件。 在这个脚本中有多个采样器在不同的主题上发布消息，目前他们正在每个采样器中创建一个生产者客户端，然后在那里关闭它。 为了提高此脚本的整体性能并使脚本更接近真实场景，我们希望在测试开始时仅初始化一次 KafkaProducer，并在测试结束时将其关闭。 所以我实施了这样的解决方案：

TestPlan
  ThreadGroup
    1.JSR223Sampler to initialize the Kafka Producer 
    2.Another JSR223 Sampler to publish a message on a topic 

使用上面初始化的相同 Kafka Producer

但问题是，当我使用 vars.putObject() 函数将 Kafka Producer 对象保存在 Sampler1 中并使用 vars.getObject 函数在 Sampler2 中获取该对象时，它会返回我 Sampler2 中的 java.lang.String 类的对象。 如果我现在尝试在 Sampler2 中投射它，它不起作用并且会抛出 ClassCastException

Sampler1 脚本：

import org.apache.kafka.clients.producer.KafkaProducer;

 Properties props = new Properties();
 props.put("bootstrap.servers", vars.get('bootstrapServer'));
 props.put("transactional.id", "my-transactional-id");
 Producer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer());
 vars.putObject("producer",producer);

Sampler2 脚本：

`KafkaProducer producer1 =  vars.getObject("producer") as KafkaProducer;
ProducerRecord<String, String> record  = new ProducerRecord<String, String>('REPO-Asset-Partner-QueryRequest', '', "{\"data\":\"{\\\"assetSearchCriteria\\\":[{\\\"attribute\\\":\\\"category\\\",\\\"operator\\\":\\\"EQ\\\",\\\"value\\\":\\\"SYSTEM\\\"}]}\"}")
record.headers().add(new RecordHeader('authorization', bear.getBytes()))
record.headers().add(new RecordHeader('ce_id', vars.get('searchIndex').getBytes()))
record.headers().add(new RecordHeader('ce_type', 'com.fico.repo.ms.asset.events.message.partner.query.AssetSearch'.getBytes()))
record.headers().add(new RecordHeader('ce_time', vars.get('time').getBytes()))
record.headers().add(new RecordHeader('requestId', vars.get('searchCorelation').getBytes()))
producer1.send(record)
producer1.close()`

例外：

Response message:javax.script.ScriptException: org.codehaus.groovy.runtime.typehandling.GroovyCastException: Cannot cast object '' with class 'java.lang.String' to class 'org.apache.kafka.clients.producer.KafkaProducer'

我仍然面临 ClassCastException 。

Answer 1

看起来你的

producer

变量是一个空字符串所以我建议检查你声明它的元素。

如果您显示的是完整代码，那么您至少缺少 StringSerializer 的 import

语句

因此检查 jmeter.log 文件 是否有任何可疑条目，并使用 Debug Sampler 检查为此线程（虚拟用户）定义的 JMeter 变量