[jira] [Updated] (KAFKA-14173) TopologyTestDriver does not perform left join on two streams when right side is missing

["Guido Josquin (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/KAFKA-14173?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Guido Josquin updated KAFKA-14173:
----------------------------------
    Description: 
I am trying to test a stream-stream join with `TopologyTestDriver`. My goal is to confirm, without running external services, that my topology performs the following left join correctly.
{code:java}
bills
  .leftJoin(payments)(
    {
      case (billValue, null) => billValue
      case (billValue, paymentValue) => (billValue.toInt - paymentValue.toInt).toString
    },
    JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMillis(100))
  )
  .to("debt")
{code}
 

In other words, if we see a `bill` and a `payment` within 100ms, the payment should be subtracted from the bill. If we do not see a payment, the debt is simply the bill.

Here is the test code.
{code:java}
val simpleLeftJoinTopology = new SimpleLeftJoinTopology
val driver = new TopologyTestDriver(simpleLeftJoinTopology.topology)
val serde = Serdes.stringSerde

val bills = driver.createInputTopic("bills", serde.serializer, serde.serializer)
val payments = driver.createInputTopic("payments", serde.serializer, serde.serializer)
val debt = driver.createOutputTopic("debt", serde.deserializer, serde.deserializer)

bills.pipeInput("fred", "100")
bills.pipeInput("george", "20")
payments.pipeInput("fred", "95")

// When in doubt, sleep twice
driver.advanceWallClockTime(Duration.ofMillis(500))
Thread.sleep(500)

val keyValues = debt.readKeyValuesToList()
keyValues should contain theSameElementsAs Seq(
  // This record is present
  new KeyValue[String, String]("fred", "5"),
  // This record is missing
  new KeyValue[String, String]("george", "20")
)
{code}
Full code available at [https://github.com/Oduig/kstreams-left-join-example]

Is seems that advancing the wall clock time, or sleeping for that matter, never triggers the join condition when data only arrives on the left side.

  was:
I am trying to test a stream-stream join with `TopologyTestDriver`. My goal is to confirm, without running external services, that my topology performs the following left join correctly.

```
bills
  .leftJoin(payments)(
    {
      case (billValue, null) => billValue
      case (billValue, paymentValue) => (billValue.toInt - paymentValue.toInt).toString
    },
    JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMillis(100))
  )
  .to("debt")
```

In other words, if we see a `bill` and a `payment` within 100ms, the payment should be subtracted from the bill. If we do not see a payment, the debt is simply the bill.

Here is the test code.

```
val simpleLeftJoinTopology = new SimpleLeftJoinTopology
val driver = new TopologyTestDriver(simpleLeftJoinTopology.topology)
val serde = Serdes.stringSerde

val bills = driver.createInputTopic("bills", serde.serializer, serde.serializer)
val payments = driver.createInputTopic("payments", serde.serializer, serde.serializer)
val debt = driver.createOutputTopic("debt", serde.deserializer, serde.deserializer)

bills.pipeInput("fred", "100")
bills.pipeInput("george", "20")
payments.pipeInput("fred", "95")

// When in doubt, sleep twice
driver.advanceWallClockTime(Duration.ofMillis(500))
Thread.sleep(500)

val keyValues = debt.readKeyValuesToList()
keyValues should contain theSameElementsAs Seq(
  // This record is present
  new KeyValue[String, String]("fred", "5"),
  // This record is missing
  new KeyValue[String, String]("george", "20")
)
```

Full code available at https://github.com/Oduig/kstreams-left-join-example

Is seems that advancing the wall clock time, or sleeping for that matter, never triggers the join condition when data only arrives on the left side.


> TopologyTestDriver does not perform left join on two streams when right side is missing
> ---------------------------------------------------------------------------------------
>
>                 Key: KAFKA-14173
>                 URL: https://issues.apache.org/jira/browse/KAFKA-14173
>             Project: Kafka
>          Issue Type: Bug
>          Components: streams-test-utils
>    Affects Versions: 2.3.1
>            Reporter: Guido Josquin
>            Priority: Minor
>
> I am trying to test a stream-stream join with `TopologyTestDriver`. My goal is to confirm, without running external services, that my topology performs the following left join correctly.
> {code:java}
> bills
>   .leftJoin(payments)(
>     {
>       case (billValue, null) => billValue
>       case (billValue, paymentValue) => (billValue.toInt - paymentValue.toInt).toString
>     },
>     JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMillis(100))
>   )
>   .to("debt")
> {code}
>  
> In other words, if we see a `bill` and a `payment` within 100ms, the payment should be subtracted from the bill. If we do not see a payment, the debt is simply the bill.
> Here is the test code.
> {code:java}
> val simpleLeftJoinTopology = new SimpleLeftJoinTopology
> val driver = new TopologyTestDriver(simpleLeftJoinTopology.topology)
> val serde = Serdes.stringSerde
> val bills = driver.createInputTopic("bills", serde.serializer, serde.serializer)
> val payments = driver.createInputTopic("payments", serde.serializer, serde.serializer)
> val debt = driver.createOutputTopic("debt", serde.deserializer, serde.deserializer)
> bills.pipeInput("fred", "100")
> bills.pipeInput("george", "20")
> payments.pipeInput("fred", "95")
> // When in doubt, sleep twice
> driver.advanceWallClockTime(Duration.ofMillis(500))
> Thread.sleep(500)
> val keyValues = debt.readKeyValuesToList()
> keyValues should contain theSameElementsAs Seq(
>   // This record is present
>   new KeyValue[String, String]("fred", "5"),
>   // This record is missing
>   new KeyValue[String, String]("george", "20")
> )
> {code}
> Full code available at [https://github.com/Oduig/kstreams-left-join-example]
> Is seems that advancing the wall clock time, or sleeping for that matter, never triggers the join condition when data only arrives on the left side.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)