Does LeftJoin produce a message for main steam before message appeared for the other stream

[ppilev]

hello.

I have a left join where the left stream can have a message (L-message) every 5 seconds and the right one has a message (R-message) every 2 seconds. also the right stream message can be published with some delay so I put grace period: 10 sec
now I'd like to have the L-message joined to the nearest R-message, so I have configured a window as follow:

beforeMs: 0 ms
afterMs: 2000 ms
graceMs: 10000 ms
maintainDurationMs: DEFAULT_RETENTION_MS = 24 * 60 * 60 * 1000L // one day

my expectation was that there will be two messages in the output topic - one for the L-message and one more when the R-message appeared but I can only see the final message produced out of the both streams.
so is it possible to get the output generated for L-stream message ASAP and later have an update when the R-message joined?

Thanks!

[LGouellec]

Hi @ppilev ,

Can you enable debug logging, reprocess the records with your expectation and push these logs here please ?

Best regards,

[ppilev]

I've investigated further and checked the code where join records are processed here
https://github.com/LGouellec/kafka-streams-dotnet/blob/8ed8faf137a1229583b0b4718afd138b300ad4bd/core/Processors/KStreamKStreamJoinProcessor.cs#L41

I think beforeMs and afterMs values should be swapped when outer is true as follows

            var backwardMs = outer ? beforeMs : afterMs;
            var forwardMs = outer ? afterMs : beforeMs;

            long timeFrom = Math.Max(0L, inputRecordTimestamp - backwardMs);
            long timeTo = Math.Max(0L, inputRecordTimestamp + forwardMs);        

After modified the code to use the swapped values I ran some tests locally against a real environment and got it behaving as expected. I ran it with three different window configs

• beforeMs: 0; afterMs: 2000
• beforeMs: 0; afterMs: 3000
• beforeMs: 1000; afterMs: 2000

all behaved as I expected and below are shown the results I got with the first config:

beforeMs: 0; afterMs: 2000
--------------------------------------
| ts  | product | detail | LEFT-JOIN |
|------------------------------------|
|0    |    x    |        |  x - null |
|------------------------------------|
|2    |         |    y   |  x - y    |
|------------------------------------|
|4    |         |    y   |   N/A     |
|------------------------------------|
|5    |    x    |        |  x - null |
|------------------------------------|
|6    |         |    y   |  x - y    |
|------------------------------------|
|8    |         |    y   |   N/A     |
--------------------------------------

Now if I'm not completely wrong about it I tried to visualize here why the values should be swapped for the outer stream.

• When stream1 is processed it's fetching records from OUTEROTHER window. That's the case when beforeMs and afterMs are fine - we're joining any record found 1 unit behind and 3 units forward its ts (the green window)
• When stream2 is processed it's fetching records from JOINTHIS window and we need swap the values in order to make the windowing correct from the perspective of stream1 which means y can be joined to any x record found 3 units behind and 1 unit forward its ts (orange window)

Best regards,

[ppilev]

hello, I made some additional research in order to confirm my expectations were correct.

I found JoinWindowOptions has documented the second stream window semantics exactly as in java docs JoinWindows

A join is symmetric in the sense, that a join specification on the first stream returns the same result record as a join specification on the second stream with flipped before and after values.

Regards,

[ppilev]

hello. I tried to crate a simplified version of that app I'm testing with but now my secondary stream data doesn't appear into joined stream at all. I have two topics - products and product-details where all the messages are published without any delay but I have ITimestampExtractor reading the timestamps from the messages which are adjusted as shown below. also all messages are using the same key.

product message - ts: x + 0 sec
detail message - ts: x + 2 sec
detail message - ts: x + 4 sec

product message - ts: x + 5 sec
detail message - ts: x + 6 sec
detail message - ts: x + 8 sec

product message - ts: x + 10 sec
detail message - ts: x + 10 sec
detail message - ts: x + 12 sec
detail message - ts: x + 14 sec

The C# code setting up the streams:

    var products = builder.Stream<TopicKey, Message<Product>>(
        "products",
        new JsonSerDes<TopicKey>(), new JsonSerDes<Message<Product>>(),
        new MessageTimestampExtractor<Message<Product>>(m => m.Data.LastUpdated));

    var details = builder.Stream<TopicKey, Message<ProductDetail>>(
        "product-details",
        new JsonSerDes<TopicKey>(), new JsonSerDes<Message<ProductDetail>>(),
        new MessageTimestampExtractor<Message<ProductDetail>>(m => m.Data.LastUpdated));

    var windowOptions = new CustomWindowsOptions(0, 2000L, 10000L, JoinWindowOptions.DEFAULT_RETENTION_MS);

    products
    .LeftJoin<Message<ProductDetail>, ProductWithDetail, JsonSerDes<Message<ProductDetail>>, JsonSerDes<ProductWithDetail>>(
    details,
    (productMessage, detailMessage) => new ProductWithDetail(productMessage.Data, detailMessage?.Data),
    windowOptions)
    .To<JsonSerDes<TopicKey>, JsonSerDes<ProductWithDetail>>("product-with-detail");

My stream config:

var config = new StreamConfig
{
    ApplicationId = "streamiz-processor-products",
    BootstrapServers = "localhost:29092",
    ReplicationFactor = 1,
    AutoOffsetReset = AutoOffsetReset.Earliest,
    CommitIntervalMs = 1000,
    Logger = LoggerFactory.Create(builder => builder
        .SetMinimumLevel(LogLevel.Debug)
        .AddConsole()
        .AddLog4Net(new Log4NetProviderOptions()))
};

The debug logs:
streamiz-log01.log

Thanks!

[ppilev]

@LGouellec,
In order to get it working (at least in the way I expected) I had to swap the values of beforeMs and afterMs

var windowOptions = new CustomWindowsOptions(2000L, 0L, 10000L, JoinWindowOptions.DEFAULT_RETENTION_MS);

this config however doesn't seem to satisfy the second part of the JoinWindowOptions expression

stream1.ts - before <= stream2.ts AND stream2.ts <= stream1.ts + after

[LGouellec]

Thanks @ppilev ,
I will have a look to your deep informations soon. I reach out to you asap.
Regards,

[ppilev]

I found it's enough to flip the values for the right join when created by StreamJoinBuilder. I also have written some additional tests and can submit a pull request if you don't mind.

Regards,

[LGouellec]

Hi @ppilev ,

Your analyse is brilliant ! You totally right, Kafka Streams JAVA swap the values beforeMs and afterMs here.

Sure ! If you wrote some tests, please feel free to submit PR and I will try to merge ASAP for the next release (
1.5.0)

Regards,

[ppilev]

hello. here it is.
#249

currently got it tested only with [beforeMs: 0, afterMs: 1 sec] window though.
could add more tests when have some spare time.

Regards,