[
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, sleeping, or sending an extra
record, never triggers the join condition when data only arrives on the left
side. It is possible to circumvent this by passing an explicit event time with
each test record. However, the behavior deviates from a real Kafka broker.
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.
{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.
> 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, sleeping, or sending an extra
> record, never triggers the join condition when data only arrives on the left
> side. It is possible to circumvent this by passing an explicit event time
> with each test record. However, the behavior deviates from a real Kafka
> broker.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
