Unit testing behavior, not implemented details...

[bmmptlgc]

Hi, I created a streaming app that reads messages from a multi schema topic and sends each message type to it's own topic, filtering out messages types that we don't want to stream. Here's the repo: https://github.com/bmmptlgc/TopicSplitter/tree/master

I am trying to figure out how to unit tests (not integration tests). I was hoping I could inject IKafkaSupplier into KafkaStreamingService:

private readonly IKafkaSupplier _kafkaSupplier;

public KafkaStreamingService(
    ...,
    IKafkaSupplier kafkaSupplier)
{
    ...
    _kafkaSupplier = kafkaSupplier ?? throw new ArgumentNullException(nameof(kafkaSupplier));
}

And pass it to KafkaStream:

var stream = new KafkaStream(builder.Build(), config, _kafkaSupplier);

In my Startup.cs I would add the KafkaSupplier as a Singleton:

services.AddSingleton(new DefaultKafkaClientSupplier(new KafkaLoggerAdapter(configuration), configuration));

And then in my unit test I would configure my DI by replacing the IKafkaSupplier above with MockKafkaSupplier, so my tests would run with mocks and not the real thing. I could even try using Moq to mock IKafkaSupplier.

But DefaultKafkaClientSupplier is internal. What a pity :(

If all the above was possible, I could endeavor to test the behavior of my app to simulate that a message was consumed and then assert on my mock that it "was published" (not for real cuz it's a unit test using a mock) to the correct topic name and with the correct payload. I could repeat this to test all the behaviors of my app by running the actual app and not testing implementation.

On the other end, I'm not 100% sure if this would work even if DefaultKafkaClientSupplier was public, but I would certainly love to give it a try.

I couldn't find any unit tests in https://github.com/LGouellec/kafka-streams-dotnet-samples. What testing approaches would be suggested for my application?

[LGouellec]

Hi @bmmptlgc ,

If you would like to perform unit testing, you can use the TopologyTestDriver to do that.
You have many examples here : TopologyTestDriverTests.

Today all kafka supplier are internals, mainly because MockKafkaSupplier and SyncKafkaSupplier don't implement every Kafka features.

So my recommendation is :

• Unit test -> TopologyTestDriver
• Integration test -> Real kafka cluster with TestContainer example

[bmmptlgc]

Further more, I will need to refactor to extract a BuildTopology method, which I was going to do anyways, but I was going to make that method private, as it should be. But for the sake of testing I will have to make it public. That to me is a code smell. Unfortunately, the root cause is not my own code. It is also possible for the root of the problem to be in the Confluent library itself.

I understand your point of view.

• First you can make your BuildTopology method internal instead of public and use the InternalsVisibleToAttribute if you use multiple projects.
• There is no mock consumers and producers supported, so ok I can change the visibility of the DefaultKafkaClientSupplier to public, but it won't change anything on your tests.

A few days ago I forked your repo and made DefaultKafkaClientSupplier. I was able to mock the consumer, but had another issue with a listener function set on the consumer that ends up setting a list of partitions. On my test the list is empty or null, so when I simulate that a message is received and it reaches that point it doesn't follow the path it should. But I can show it to you in more detail once I have a bit more time.

[bmmptlgc]

Can you confirm those methods don't exist in version 1.5.1? It seems to work with 1.6.0-rc2.
I added FollowMetadata = true to the stream config. and I am publishing like this:

inputTopic.PipeInput("test-1", order, Headers);

But in my topology filter StreamizMetadata.GetCurrentHeadersMetadata() is returning null 😞
Any idea what I might be missing?

Yes 1.5.1 use Confluent.SchemaRegistry - 2.2.0 which not support GetCompatibilityAsync. Streamiz 1.6.X support it.

@bmmptlgc What parameter mode have you passed into the TopologyTestDriver ctor ? SYNC_TASK. or ASYNC_CLUSTER_IN_MEMORY ? If you have set ASYNC_CLUSTER_IN_MEMORY, Headers are not correctly mock today. Feel free to create a GH issue to track this lack. SYNC_TASK does it normally.

I used the default SYNC_TASK and didn't get any headers, so I changed to the ASYNC_CLUSTER_IN_MEMORY and no headers either.

Example code here: https://github.com/bmmptlgc/kafka-streams-dotnet/tree/kafka-supplier-di

You can set a breakpoint on line 65 of https://github.com/bmmptlgc/kafka-streams-dotnet/blob/kafka-supplier-di/sample-kafka-supplier-di/TopicSplitterService.cs
Then run the test on line 46 of https://github.com/bmmptlgc/kafka-streams-dotnet/blob/kafka-supplier-di/test/Sample.Kafka.Supplier.DI.UnitTests/TopologyDriverTests/When_consuming_a_message_with_a_type_configured_to_publish_to_a_single_schema_topic.cs
You can change the mode on line 39 of https://github.com/bmmptlgc/kafka-streams-dotnet/blob/kafka-supplier-di/test/Sample.Kafka.Supplier.DI.UnitTests/TopologyDriverTests/BaseContext.cs

[LGouellec]

Found the bugs and fixed it in the issue/338 branch

[bmmptlgc]

Further more, I will need to refactor to extract a BuildTopology method, which I was going to do anyways, but I was going to make that method private, as it should be. But for the sake of testing I will have to make it public. That to me is a code smell. Unfortunately, the root cause is not my own code. It is also possible for the root of the problem to be in the Confluent library itself.

I understand your point of view.

• First you can make your BuildTopology method internal instead of public and use the InternalsVisibleToAttribute if you use multiple projects.
• There is no mock consumers and producers supported, so ok I can change the visibility of the DefaultKafkaClientSupplier to public, but it won't change anything on your tests.

A few days ago I forked your repo and made DefaultKafkaClientSupplier. I was able to mock the consumer, but had another issue with a listener function set on the consumer that ends up setting a list of partitions. On my test the list is empty or null, so when I simulate that a message is received and it reaches that point it doesn't follow the path it should. But I can show it to you in more detail once I have a bit more time.

So, I managed to find sometime to work on the kafka supplier Dependency Injection tests. I was able to mock the consumers and the producers just fine, but I'm having issues with the schema registry mock, since I'm not able to add any schemas.
I get an exception from Confluent GenericDesetializerImpl.cs on the line after this line:

Confluent.SchemaRegistry.Schema schema1 = await this.schemaRegistryClient.GetSchemaAsync(writerId).ConfigureAwait(false);

The schemaRegistryClient is of type MockSchemaRegistryClient and it has no schemas registered. I don't see how I can get around this,.
It would be nice if the schema registry was managed by IKafkaSupplier, with a GetSchemaRegistry method similar to GetConsumer and GetProducer. Then I would be able to use Moq or to build my own instance of MockSchemaRegistryClient with registered schemas.

You can run the test on line 7 (Debug all tests) of https://github.com/bmmptlgc/kafka-streams-dotnet/blob/kafka-supplier-di/test/Sample.Kafka.Supplier.DI.UnitTests/KafkaTopicSplitter/When_consuming_a_message_with_a_type_configured_to_publish_to_a_single_schema_topic.cs

To see the simulated message being consumed, you can put a breakpoint on line 258 of https://github.com/bmmptlgc/kafka-streams-dotnet/blob/kafka-supplier-di/core/Processors/StreamThread.cs#L258

Continuing from that point, to see the exception, you can put a breakpoint on line 325 of https://github.com/bmmptlgc/kafka-streams-dotnet/blob/kafka-supplier-di/core/Processors/AbstractProcessor.cs#L325

[bmmptlgc]

Btw, I experimented with GetSchemaRegistry on IKafkaSupplier and "hacked" a bit to pass it through and I was able to mock scehamregistry and run my tests successfully. However, I would be interested to know if there is another way around the issue described above.

[LGouellec]

So, I managed to find sometime to work on the kafka supplier Dependency Injection tests. I was able to mock the consumers and the producers just fine, but I'm having issues with the schema registry mock, since I'm not able to add any schemas.
I get an exception from Confluent GenericDesetializerImpl.cs on the line after this line

What exception do you have ? Can you post the stack trace ?
When I run your unit test locally, it seems your tests never ended, and I don't have exception at all ..

Btw, I experimented with GetSchemaRegistry on IKafkaSupplier and "hacked" a bit to pass it through and I was able to mock scehamregistry and run my tests successfully. However, I would be interested to know if there is another way around the issue described above.

IKafkaSupplier is part of the Streamiz Core project, and I'd like to keep this project without big dependencies. In the mean time, the KafkaSupplier is responsible to build Consumer, Producer, AdminClient, every Kafka clients where the Schema Registry client is part of the client and producer to serialize or deserialize the records.
Futhermore, you can use Streamiz and all their stuff without schemas, so that's why the Core project doesn't have dependency of Schema Registry.

[bmmptlgc]

I was more worried about getting the mocking to work than the actual flow of the tests. There is no assertion in my test (yet), and the exception is being handled and the CompleteShutdown is executed, so the test passes. But if you place the breakpoints I mentioned and debug there is a Null Reference exception set as an inner exception to a DeserializationException, online 329 of AbstractProcessor.cs.

[LGouellec]

When I debug the test located here : https://github.com/bmmptlgc/kafka-streams-dotnet/blob/6cb78e946a4b87d4b78763305b7c55d7e7f55e18/test/Sample.Kafka.Supplier.DI.UnitTests/KafkaTopicSplitter/When_consuming_a_message_with_a_type_configured_to_publish_to_a_single_schema_topic.cs#L29 with IDE (Rider), I don't have exception.

Same if I run your unit test with the dotnet test cli, nothing just a start of output and nothing.

Can you try to write a unit test with a single class and the TopologyTestDriver and your mock of consumer to reproduce easier the issue ?

[bmmptlgc]

I just merged develop into my branch, to make sure I wasn't missing any recent changes and I still get the same exception. I will try the TopologyTestDriver test you mentioned, but that will likely not run the same code.

Just to make it clear, I get the exception running the KafkaTopicSplitter folder (where I don't use the test driver/harness) and not the TopologyDriverTests folder.

Also, there is some recent change with Rider that it is not running/debugging tests properly for my NameAttribute.cs override. I have to debug from the class for it to work:
https://github.com/bmmptlgc/kafka-streams-dotnet/blob/6cb78e946a4b87d4b78763305b7c55d7e7f55e18/test/Sample.Kafka.Supplier.DI.UnitTests/KafkaTopicSplitter/When_consuming_a_message_with_a_type_configured_to_publish_to_a_single_schema_topic.cs#L7

In the mean time (while I try you suggested test with TopologyTestDriver), could you debug my test again (as shown above) and put breakpoints in Confluent's GenericDeserializerImpl.cs lines 56 and 57 and tell me what value you have in this.schemaRegistryClient when stoppped on line 56 and then schema1 after moving to line 57?

[bmmptlgc]

I finally nailed it! And only by changing DefaultKafkaClientSupplier , DefaultKafkaClientBuilder and KafkaLoggerAdapter from public to internal.

Here is how I solved my schema registry issue:

• I figured that the test driver tests registered the schema (schemaRegistryClient.RegisterSchemaAsync) when I simulated the consumption of a message (inputTopic.PipeInput("test-1", order, Headers)), since this causes a serialization and that's when the Confluent lib is making that call to register the schema. Then when the deserializer calls schemaRegistryClient.GetSchemaAsync and the schema is there, so it doesn't throw null after.
• In my dependency injection type tests, I mock the consumer and fake the consumption of a message, so no serialization takes place and then the schema is not registered.
• I also figured that the schema registry client is being set on the DefaultValueSerDes which is a property I can access on my initial StreamConfig using reflection. So, when setting the behavior for my IConsumer mock I used reflection to call schemaRegistryClient.RegisterSchemaAsync and it works beautifully.

I am also stopping my hosted service now, so that the streaming service doesn't keep running in the background after my test assertion.

So, yeah, I would love if you didn't mind making those classes public 😄

[LGouellec]

Thanks for you analysis,

Will do in the next release