Stream record context timestamp

[ppilev]

hi @LGouellec

I have the following code (pseudo code below) creating tempStream which later is used in windowed joins
var tempStream = stream.GroupBy.Aggregate.ToStream.To(TopicTimestampExtractor<K, V>)

here my TopicTimestampExtractor is modifying the timestamp of the record context.
I found there was submitted a pull request for this "feature" and now recordContext.ChangeTimestamp can be used to update records timestamp.
and all this works just fine

internal class TopicTimestampExtractor<K, V> : ITopicNameExtractor<K, V>
{
    private readonly string topicName;
    private readonly Func<V, DateTime> extractor;

    public TopicTimestampExtractor(string topicName, Func<V, DateTime> extractor)
    {
        this.topicName = topicName;
        this.extractor = extractor;
    }
    public string Extract(K key, V value, IRecordContext recordContext)
    {
        var datetime = extractor(value);
        recordContext.ChangeTimestamp(datetime.GetMilliseconds());
        return topicName;
    }
}

now because I don't need the data of that tempStream to be published in kafka I have removed the materialization To(TopicTimestampExtractor<K, V>) but now the windowed joins don't work anymore.
I suspect that's due to record timestamp.

I checked if I can use Transform method with custom ITransformer<> but couldn't find how to create a new Record with custom timestamp as the Timestamp is internal property and also the constructors are internal.

Any idea about how to modify the record timestamps without doing materialization?

Thanks,

[LGouellec]

Hi @ppilev,

For now, you can't unfortunately. I have to switch this method to public :
https://github.com/LGouellec/kafka-streams-dotnet/blob/a6189c8882e1220c82e9f3931bb38d360eb5156c/core/ProcessorContext.cs#L105

Could you create a pull request, I will merge ASAP and be available in 1.5.0 release ?

Many thanks

[ppilev]

hello, I've submitted this pull request for it #255
it's using a new graph node and processor for it.
I know that's not exactly what you asked for but could you please review it?

thanks,