[jira] [Created] (KAFKA-12542) TopologyTestDriver

["SHWETA SINHA (Jira)" <ji...@apache.org>]

SHWETA SINHA created KAFKA-12542:
------------------------------------

             Summary: TopologyTestDriver
                 Key: KAFKA-12542
                 URL: https://issues.apache.org/jira/browse/KAFKA-12542
             Project: Kafka
          Issue Type: Bug
          Components: streams-test-utils
    Affects Versions: 2.6.0
            Reporter: SHWETA SINHA


I am using Kafka Streams DSL to create topology.

StreamsBuilder streamsBuilder=new StreamsBuilder();
valueSerde.configure(getConfigForSpecificAvro(appId),false);
KStream<String, AvroDTO> stream = streamsBuilder.stream(inputTopic, Consumed.with(new Serdes.StringSerde(), valueSerde));
KStream<String, AvroDTO> filtered = stream .filter((key, value) -> ServiceConsumer.filter(key,value));
filtered .map((KeyValueMapper<String, AvroDTO, KeyValue<String, SpecificRecordBase>>) (key, value) -> ServiceConsumer.process(key,value))
 .to((k,v,recordContext) -> v instanceof AvroDTO? dlqTopic:outputTopic,Produced.with(new Serdes.StringSerde(), valueSerde));
Topology topology=streamsBuilder.build();
KafkaStreams kafkaStreams=new KafkaStreams(topology,getKafkaStreamsConfig(appId));

 

To Test the Topology, I am using TopologyTestDriver.  

when(getKafkaStreamsConfig(any())).thenReturn(kafkaConfig);
 when(ServiceConsumer.filter(any(),any())).thenReturn(false);
// when(ServiceConsumer.process(any(),any())).thenReturn(new KeyValue<>(statusDto.getTaskId().toString(),statusDto));
 topologyTestDriver=new TopologyTestDriver(getTopologyAndStartKafkaStreams(),kafkaConfig);
StreamInput = topologyTestDriver.createInputTopic("INPUT_TOPIC", new StringSerializer(), new KafkaAvroSerializer(schemaRegistryClient));
UpdateOutput =topologyTestDriver.createOutputTopic("OUTPUT_TOPIC",new StringDeserializer(),new KafkaAvroDeserializer(schemaRegistryClient,config));

StreamInput.pipeInput("Hi);
assertThat(UpdateOutput .isEmpty()).isTrue();

 

I am checking if there are no filtered messages then my output topic is empty.

Getting Error while Unit Testing

java.lang.IllegalArgumentException: Unknown topic: OUTPUT_TOPIC

 

Changing when(ServiceConsumer.filter(any(),any())).thenReturn(false); to true doesnt gives any error.

 



--
This message was sent by Atlassian Jira
(v8.3.4#803005)