Need help with building custom KeyValueStoreBuilder

[shashi812]

I have a requirement to pull few properties from the value part of topic message and build state store with custom key and value.

Ex: Value : studentId, age, Name, cityId.

I want state store to store results with Key: studentId, cityId and Value : studentId, age, Name, cityId.

code for KeyValueStoreBuilder:

_builder.AddStateStore(Stores.KeyValueStoreBuilder(Stores.PersistentKeyValueStore(storeName), subscriptionKeySpecificAvroSerde, subscriptionSpecificAvroSerde));

code for stream and transform method

`var stream = _builder.Stream(topicName,
	subscriptionKeySpecificAvroSerde, subscriptionSpecificAvroSerde);
stream.Transform(new TransformerSupplier(), "output");`

I need help with creation of TransformerSupplier.

I learned that we need to create a transformer like below

`private class Transformer : ITransformer<K, V, K1, V1>
{
private IKeyValueStore<K1, V1> store;

public void Close()
{
	throw new NotImplementedException();
}

public void Init(ProcessorContext<K1, V1> context)
{
	// Ensure the state store name matches the one used in the builder
	store = (IKeyValueStore<K1, V1>)context.GetStateStore(storeName);
	if (store == null)
	{
		throw new InvalidOperationException("State store  not found");
	}
}

public Record<K1, V1> Process(Record<K1, V1> record)
{
	var key = new K1
	{
		StudentId = record.Value.StudentId,
		CityId = record.Value.CityId
	};

	var storeValue = store.Get(key);
	store.Put(key, storeValue);

	return Record<K1, V1>.Create(key, storeValue);
}

}`
But the Transform or TransformValues is expecting TransformerSupplier

Stack Trace:
this usually indicate Streams internal errors
System.NullReferenceException: Object reference not set to an instance of an object.
at Streamiz.Kafka.Net.Processors.KStreamTransformer4.Init(ProcessorContext context) at Streamiz.Kafka.Net.Processors.AbstractProcessor2.Init(ProcessorContext context)
at Streamiz.Kafka.Net.Processors.SourceProcessor2.Init(ProcessorContext context) at Streamiz.Kafka.Net.Processors.StreamTask.InitializeTopology() at Streamiz.Kafka.Net.Processors.Internal.TaskManager.CreateTasks(ICollection1 assignment)
at Streamiz.Kafka.Net.Kafka.Internal.StreamsRebalanceListener.PartitionsAssigned(IConsumer2 consumer, List1 partitions)

[LGouellec]

Hey @shashi812 ,

Please find an example here :

streamiz/test/Streamiz.Kafka.Net.Tests/Processors/KStreamTransformerAPITests.cs

Line 139 in 509076f

builder.Stream<string, string>("topic")

Let me know if you have any further questions.

Kr,

[shashi812]

Thank you @LGouellec, I have two more questions

1. How to read from data from state store only after the Transformer finished processing of all the records.
2. How to filter out records based on some criteria in the process method of Transformer

[LGouellec]

How to read from data from state store only after the Transformer finished processing of all the records.

it's not clear, the Transformer process data record by record, you can get access from the state store directly. I don't understand what do you mean by finished processing of all the records ?

2- If the process method return null, the record is ignored and filtered out the next processor