diff options
author | 2023-03-29 12:52:11 +0100 | |
---|---|---|
committer | 2023-04-03 09:48:00 +0100 | |
commit | c3386ebc4fd444e810d8739f7ddd78765d36c631 (patch) | |
tree | eee955a759be4cf85d6052b56f1ac83ae75c851b /sampleClient/src/main/java/org | |
parent | 2d3181aa8e21a6f4b31565958cf2cb35bc836dff (diff) |
Signed-off-by: efiacor <fiachra.corcoran@est.tech>
Change-Id: Iaa3c2ff9b60aa5a691242deec82911a700222d9a
Issue-ID: DMAAP-1847
Diffstat (limited to 'sampleClient/src/main/java/org')
-rw-r--r-- | sampleClient/src/main/java/org/onap/dmaap/kafka/sample/Main.java | 24 | ||||
-rw-r--r-- | sampleClient/src/main/java/org/onap/dmaap/kafka/sample/SampleConfiguration.java | 1 |
2 files changed, 22 insertions, 3 deletions
diff --git a/sampleClient/src/main/java/org/onap/dmaap/kafka/sample/Main.java b/sampleClient/src/main/java/org/onap/dmaap/kafka/sample/Main.java index 37a3097..c80c87f 100644 --- a/sampleClient/src/main/java/org/onap/dmaap/kafka/sample/Main.java +++ b/sampleClient/src/main/java/org/onap/dmaap/kafka/sample/Main.java @@ -20,7 +20,11 @@ package org.onap.dmaap.kafka.sample; +import java.util.List; +import org.apache.kafka.clients.producer.RecordMetadata; import org.onap.dmaap.kafka.OnapKafkaClient; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.CommandLineRunner; import org.springframework.boot.SpringApplication; @@ -29,6 +33,8 @@ import org.springframework.boot.autoconfigure.SpringBootApplication; @SpringBootApplication public class Main implements CommandLineRunner{ + private final Logger log = LoggerFactory.getLogger(OnapKafkaClient.class.getName()); + @Autowired private SampleConfiguration configuration; @@ -37,8 +43,22 @@ public class Main implements CommandLineRunner{ } @Override - public void run(String... args) { + public void run(String... args) throws InterruptedException { OnapKafkaClient handler = new OnapKafkaClient(configuration); - handler.fetchFromTopic("dummy.topic.blah"); + String testTopic = configuration.getConsumerTopics().get(0); + for (int i = 0; i < 5; i++) { + RecordMetadata recordMetadata = handler.publishToTopic(testTopic, "dummy-message-"+i); + if (recordMetadata != null) { + log.info("Topic: {}, Partition: {}, Offset: {}", recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset()); + } + } + int fetch = 0; + while (true) { + fetch++; + log.info("Fetch {} from topic: {}", fetch, testTopic); + List<String> res = handler.fetchFromTopic(testTopic); + log.info("Messages from fetch {}: " + res, fetch); + Thread.sleep(3000); + } } }
\ No newline at end of file diff --git a/sampleClient/src/main/java/org/onap/dmaap/kafka/sample/SampleConfiguration.java b/sampleClient/src/main/java/org/onap/dmaap/kafka/sample/SampleConfiguration.java index 0cb5498..601504a 100644 --- a/sampleClient/src/main/java/org/onap/dmaap/kafka/sample/SampleConfiguration.java +++ b/sampleClient/src/main/java/org/onap/dmaap/kafka/sample/SampleConfiguration.java @@ -44,5 +44,4 @@ public class SampleConfiguration implements IKafkaConfig { private List<String> producerTopics; private String kafkaSaslJaasConfig; - // private String kafkaSaslMechanism; } |