[GitHub] [kafka] vamossagar12 commented on pull request #12756: Kafka 12960: Follow up Commit to filter expired records from Windowed/Session Stores

[GitBox <gi...@apache.org>]

vamossagar12 commented on PR #12756:
URL: https://github.com/apache/kafka/pull/12756#issuecomment-1279768865

   @vpapavas I need your help on this test failure: org.apache.kafka.streams.kstream.internals.KStreamKStreamSelfJoinTest.shouldMatchInnerJoinWithSelfJoinOutOfOrderMessages failing with the following stacktrace:
   
   ```
   java.lang.AssertionError: the number of outputs:[Record{key=A, value=11, timestamp=0, headers=RecordHeaders(headers = [], isReadOnly = false)}, Record{key=A, value=21, timestamp=9999, headers=RecordHeaders(headers = [], isReadOnly = false)}, Record{key=A, value=12, timestamp=9999, headers=RecordHeaders(headers = [], isReadOnly = false)}, Record{key=A, value=22, timestamp=9999, headers=RecordHeaders(headers = [], isReadOnly = false)}, Record{key=B, value=11, timestamp=11000, headers=RecordHeaders(headers = [], isReadOnly = false)}, Record{key=A, value=32, timestamp=13000, headers=RecordHeaders(headers = [], isReadOnly = false)}, Record{key=A, value=23, timestamp=13000, headers=RecordHeaders(headers = [], isReadOnly = false)}, Record{key=A, value=33, timestamp=13000, headers=RecordHeaders(headers = [], isReadOnly = false)}, Record{key=A, value=42, timestamp=15000, headers=RecordHeaders(headers = [], isReadOnly = false)}, Record{key=A, value=43, timestamp=15000, headers=RecordHeaders
 (headers = [], isReadOnly = false)}, Record{key=A, value=24, timestamp=15000, headers=RecordHeaders(headers = [], isReadOnly = false)}, Record{key=A, value=34, timestamp=15000, headers=RecordHeaders(headers = [], isReadOnly = false)}, Record{key=A, value=44, timestamp=15000, headers=RecordHeaders(headers = [], isReadOnly = false)}, Record{key=C, value=11, timestamp=16000, headers=RecordHeaders(headers = [], isReadOnly = false)}, Record{key=D, value=11, timestamp=17000, headers=RecordHeaders(headers = [], isReadOnly = false)}, Record{key=A, value=55, timestamp=30000, headers=RecordHeaders(headers = [], isReadOnly = false)}, Record{key=A, value=53, timestamp=13000, headers=RecordHeaders(headers = [], isReadOnly = false)}, Record{key=A, value=54, timestamp=15000, headers=RecordHeaders(headers = [], isReadOnly = false)}, Record{key=A, value=55, timestamp=6000, headers=RecordHeaders(headers = [], isReadOnly = false)}, Record{key=A, value=35, timestamp=13000, headers=RecordHeaders(headers
  = [], isReadOnly = false)}, Record{key=A, value=45, timestamp=15000, headers=RecordHeaders(headers = [], isReadOnly = false)}]
   Expected: is <20>
        but: was <21>
   	at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:20)
   	at org.apache.kafka.test.MockApiProcessor.checkAndClearProcessResult(MockApiProcessor.java:97)
   	at org.apache.kafka.streams.kstream.internals.KStreamKStreamSelfJoinTest.shouldMatchInnerJoinWithSelfJoinOutOfOrderMessages(KStreamKStreamSelfJoinTest.java:334)
   	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
   	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
   	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
   	at java.lang.reflect.Method.invoke(Method.java:498)
   	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59)
   	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
   	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56)
   	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
   	at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306)
   	at org.junit.runners.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100)
   	at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:366)
   	at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:103)
   	at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:63)
   	at org.junit.runners.ParentRunner$4.run(ParentRunner.java:331)
   	at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79)
   	at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:329)
   	at org.junit.runners.ParentRunner.access$100(ParentRunner.java:66)
   	at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:293)
   	at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306)
   	at org.junit.runners.ParentRunner.run(ParentRunner.java:413)
   	at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
   	at org.junit.runner.JUnitCore.run(JUnitCore.java:115)
   	at org.junit.vintage.engine.execution.RunnerExecutor.execute(RunnerExecutor.java:42)
   	at org.junit.vintage.engine.VintageTestEngine.executeAllChildren(VintageTestEngine.java:80)
   	at org.junit.vintage.engine.VintageTestEngine.execute(VintageTestEngine.java:72)
   	at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:107)
   	at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:88)
   	at org.junit.platform.launcher.core.EngineExecutionOrchestrator.lambda$execute$0(EngineExecutionOrchestrator.java:54)
   	at org.junit.platform.launcher.core.EngineExecutionOrchestrator.withInterceptedStreams(EngineExecutionOrchestrator.java:67)
   	at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:52)
   	at org.junit.platform.launcher.core.DefaultLauncher.execute(DefaultLauncher.java:114)
   	at org.junit.platform.launcher.core.DefaultLauncher.execute(DefaultLauncher.java:86)
   	at org.junit.platform.launcher.core.DefaultLauncherSession$DelegatingLauncher.execute(DefaultLauncherSession.java:86)
   	at org.junit.platform.launcher.core.SessionPerRequestLauncher.execute(SessionPerRequestLauncher.java:53)
   	at org.gradle.api.internal.tasks.testing.junitplatform.JUnitPlatformTestClassProcessor$CollectAllTestClassesExecutor.processAllTestClasses(JUnitPlatformTestClassProcessor.java:99)
   	at org.gradle.api.internal.tasks.testing.junitplatform.JUnitPlatformTestClassProcessor$CollectAllTestClassesExecutor.access$000(JUnitPlatformTestClassProcessor.java:79)
   	at org.gradle.api.internal.tasks.testing.junitplatform.JUnitPlatformTestClassProcessor.stop(JUnitPlatformTestClassProcessor.java:75)
   	at org.gradle.api.internal.tasks.testing.SuiteTestClassProcessor.stop(SuiteTestClassProcessor.java:61)
   	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
   	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
   	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
   	at java.lang.reflect.Method.invoke(Method.java:498)
   	at org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:36)
   	at org.gradle.internal.dispatc…
   ```
   
   This PR is a follow up to https://github.com/apache/kafka/pull/11211 and it aims to filter expired records from stateful persistent stores(in-memory stores already supports this). 11211 was merged recently and rolled back because it broke a couple of test and one of them is this. I see that your changes were introduced around the same time I made the last commit on the PR. Either ways, looks like where I need your help is this bit of code:
   
   ```
   // This is needed so that output records follow timestamp order
                       // Join this with self
                       if (inputRecordTimestamp < maxRecordTimestamp && !emittedJoinWithSelf) {
                           emittedJoinWithSelf = true;
                           context().forward(selfRecord);
                       }
   ```
   
   To give you more context, what's happening is that due to the expiry of records, the Inner join is not returning the self-join records. The stores when doing the fetch call, compute the lower bound by using this formula:
   
   ```
   return Math.max(from, observedStreamTime - retentionPeriod + 1);
   ```
   
   due to which the Self join records are ignored from inner join. But, because of the condition above, the self join calculation is returning the self join record causing the test to fail.
   
   What I want to know is when records are being expired, with this logic there's a chance that the output of inner join and self-join for out of order records might not match. The way I see it, what inner join is returning now seems correct. WDYT? cc @ableegoldman , @cadonna 
   
   
    


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org