Forum Discussion

amirse's avatar
amirse
Contributor
4 years ago

ReadyAPI - Adding assertions on Received Data from Kafka subscribe event

I am using readyapi to read some events from our Kafka queue. I have successfully connected, Subcribed, to the queue and I see events coming in while running some tests which creates events on that t...
Assertions
TNeuschwanger's avatar
TNeuschwanger
Champion Level 1

Hello amirse 

I don't know the answer, and it has been a long time ago that I dealt with messaging systems... 

 

Things that I remember biting me were:

-time to life of the message.  Could that have expired in your case and it is therefore no longer available on the queue or topic for the GetData to see?

-read once then delete.  Could that have happened in your case and it is therefore no longer available on the queue or topic for the GetData to see?

 

... just throwing out a couple of things to think about.

 

Sorry,

Todd

amirse's avatar
amirse
Contributor
4 years ago

Hi,

thank you for your suggestions. I have managed to modify the groovy script found in the below ticket to get data

https://community.smartbear.com/t5/SoapUI-Open-Source/Need-groovy-script-to-read-messages-from-kafka-topic/td-p/196019

 

However I really hope smartbear will come back to us with a solution on how to get that response with the "GetData" functionality. 

 

Until they do here is the code I have modified that now works for me. 

Hope it will help other people

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

log.info("*************************************************************************")
log.info("                                      TOPIC NAME    ")
log.info("*************************************************************************")


//SAVE RECEIVED DATA
def tsuite1 = testRunner.testCase.testSuite.project.getTestSuiteByName("TEST SUITE NAME");
def tcase1= tsuite1.getTestCaseByName("TEST CASE NAME");
db = tcase1.testSteps['Data Source'].dataSource.gridModel
int row =  testRunner.testCase.testSteps['Data Source'].currentRow
def guid = "ABC12345DEF"


//KAFKA SETTINGS
String topic = "MY-TOPIC-NAME"
Properties properties = new Properties();
properties.put("group.id", "test");
properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.put("security.protocol","SSL");
properties.put("ssl.keystore.password", "XXX");
properties.put("bootstrap.servers", "XXX:6667");
properties.put("ssl.key.password", "XXX");
properties.put("ssl.truststore.password", "XXX");
properties.put("ssl.truststore.location", "C:\\MyCertificates\\certificate.jks");
properties.put("ssl.keystore.location", "C:\\MyCertificates\\certificate.jks");
//properties.put("session.timeout.ms", "10000");
properties.put("max.poll.records", "5");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
consumer.subscribe(Arrays.asList(topic));
long t = System.currentTimeMillis();
long end = t + 9000;
while (System.currentTimeMillis()<end){
	log.info("START READING")
	ConsumerRecords<String, String> records = consumer.poll(200);
	for (ConsumerRecord<String, String> record : records){
		log.info("key = " + record.key())
		log.info("value = " + record.value());

		if(record.value().contains(guid)){
			log.info("found data")
			db.setValueAt(topic, row, 1);
			db.setValueAt(guid, row, 2);
		}
	}
	log.info("STOP READING")
}
  consumer.close();