{"action":"created","issue":{"url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/issues/487","repository_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK","labels_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/issues/487/labels{/name}","comments_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/issues/487/comments","events_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/issues/487/events","html_url":"https://github.com/GoogleCloudPlatform/DataflowJavaSDK/issues/487","id":190120955,"number":487,"title":"Odd behaviour of SlidingWindows when used with TestPipeline","user":{"login":"dkruchinin","id":19095,"avatar_url":"https://avatars.githubusercontent.com/u/19095?v=3","gravatar_id":"","url":"https://api.github.com/users/dkruchinin","html_url":"https://github.com/dkruchinin","followers_url":"https://api.github.com/users/dkruchinin/followers","following_url":"https://api.github.com/users/dkruchinin/following{/other_user}","gists_url":"https://api.github.com/users/dkruchinin/gists{/gist_id}","starred_url":"https://api.github.com/users/dkruchinin/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dkruchinin/subscriptions","organizations_url":"https://api.github.com/users/dkruchinin/orgs","repos_url":"https://api.github.com/users/dkruchinin/repos","events_url":"https://api.github.com/users/dkruchinin/events{/privacy}","received_events_url":"https://api.github.com/users/dkruchinin/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":1,"created_at":"2016-11-17T18:07:03Z","updated_at":"2016-11-17T19:01:32Z","closed_at":null,"body":"I've got a simple test that demonstrate an odd behaviour of sliding window when used with TestPipeline. Basically a bunch of strings is fed to the input, then they get accumulated in the sliding window, then the sum aggregation is applied to count the duplicates and finally the output of the aggregation function is logged. With a sliding window of 10 minutes duration and 5 minutes period the pipeline should produce two intersecting windows: [-5, 5] and [0, 10]...\r\n\r\n```java\r\npublic class SlidingWindowTest {\r\n    private static PipelineOptions options = PipelineOptionsFactory.create();\r\n    private static final Logger LOG = LoggerFactory.getLogger(SlidingWindowTest.class);\r\n\r\n    private static class IdentityDoFn extends DoFn<KV<String, Integer>, KV<String, Integer>>\r\n        implements DoFn.RequiresWindowAccess{\r\n\r\n        @Override\r\n        public void processElement(ProcessContext processContext) throws Exception {\r\n            KV<String, Integer> item = processContext.element();\r\n            LOG.info(\"~~~~~~~~~~> {} => {}\", item.getKey(), item.getValue());\r\n            LOG.info(\"~~~~~~~~~~~ {}\", processContext.window());\r\n            processContext.output(item);\r\n        }\r\n    }\r\n\r\n    private static class OutputWithCurrentTsFn extends DoFn<String, String> {\r\n\r\n        @Override\r\n        public void processElement(ProcessContext processContext) throws Exception {\r\n            processContext.outputWithTimestamp(processContext.element(), Instant.now());\r\n        }\r\n    }\r\n\r\n    @Test\r\n    public void whatsWrongWithSlidingWindow() {\r\n        Pipeline p = TestPipeline.create(options);\r\n\r\n        p.apply(Create.of(\"cab\", \"abc\", \"a1b2c3\", \"abc\", \"a1b2c3\"))\r\n            //.apply(ParDo.of(new OutputWithCurrentTsFn()))\r\n            .apply(MapElements.via((String item) -> KV.of(item, 1))\r\n                       .withOutputType(new TypeDescriptor<KV<String, Integer>>() {}))\r\n            .apply(Window.<KV<String, Integer>>into(SlidingWindows.of(Duration.standardMinutes(10))\r\n                                                        .every(Duration.standardMinutes(5))))\r\n            .apply(Sum.integersPerKey())\r\n            .apply(ParDo.of(new IdentityDoFn()));\r\n\r\n        p.run();\r\n    }\r\n}\r\n```\r\n\r\nbut that's what I get instead:\r\n\r\n```\r\n...\r\n17:52:00.246 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~> cab => 1\r\n17:52:00.246 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~~ [-290308-12-21T20:00:00.000Z..-290308-12-21T20:10:00.000Z)\r\n17:52:00.246 [main] DEBUG c.g.c.d.sdk.coders.CoderRegistry - Default coder for class java.lang.String found by factory\r\n17:52:00.246 [main] DEBUG c.g.c.d.sdk.coders.CoderRegistry - Default coder for class java.lang.Integer found by factory\r\n17:52:00.246 [main] DEBUG c.g.c.d.sdk.coders.CoderRegistry - Default coder for com.google.cloud.dataflow.sdk.values.KV<java.lang.String, java.lang.Integer>: KvCoder(StringUtf8Coder, VarIntCoder)\r\n17:52:00.247 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~> cab => 1\r\n17:52:00.247 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~~ [-290308-12-21T19:55:00.000Z..-290308-12-21T20:05:00.000Z)\r\n17:52:00.247 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~> abc => 2\r\n17:52:00.247 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~~ [-290308-12-21T19:55:00.000Z..-290308-12-21T20:05:00.000Z)\r\n17:52:00.247 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~> abc => 2\r\n17:52:00.247 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~~ [-290308-12-21T19:50:00.000Z..-290308-12-21T20:00:00.000Z)\r\n17:52:00.247 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~> abc => 2\r\n17:52:00.247 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~~ [-290308-12-21T20:00:00.000Z..-290308-12-21T20:10:00.000Z)\r\n17:52:00.247 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~> a1b2c3 => 2\r\n17:52:00.247 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~~ [-290308-12-21T19:50:00.000Z..-290308-12-21T20:00:00.000Z)\r\n17:52:00.248 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~> a1b2c3 => 2\r\n17:52:00.248 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~~ [-290308-12-21T20:00:00.000Z..-290308-12-21T20:10:00.000Z)\r\n17:52:00.248 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~> a1b2c3 => 2\r\n17:52:00.248 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~~ [-290308-12-21T19:55:00.000Z..-290308-12-21T20:05:00.000Z)\r\n17:52:00.249 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~> cab => 1\r\n17:52:00.249 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~~ [-290308-12-21T19:50:00.000Z..-290308-12-21T20:00:00.000Z)\r\n...\r\n```\r\n\r\nAs you can see there're three windows instead of two and all three strings get to the \"extra\" window that doesn't intersect in any way with the other two.\r\n\r\nNow the interesting thing, if you uncomment **.apply(ParDo.of(new OutputWithCurrentTsFn()))** line and run the test again, it works as expected producing only two intersecting sliding windows:\r\n\r\n```\r\n...\r\n18:01:30.904 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~> a1b2c3 => 2\r\n18:01:30.904 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~~ [2016-11-17T17:55:00.000Z..2016-11-17T18:05:00.000Z)\r\n18:01:30.905 [main] DEBUG c.g.c.d.sdk.coders.CoderRegistry - Default coder for class java.lang.String found by factory\r\n18:01:30.905 [main] DEBUG c.g.c.d.sdk.coders.CoderRegistry - Default coder for class java.lang.Integer found by factory\r\n18:01:30.905 [main] DEBUG c.g.c.d.sdk.coders.CoderRegistry - Default coder for com.google.cloud.dataflow.sdk.values.KV<java.lang.String, java.lang.Integer>: KvCoder(StringUtf8Coder, VarIntCoder)\r\n18:01:30.905 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~> a1b2c3 => 2\r\n18:01:30.905 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~~ [2016-11-17T18:00:00.000Z..2016-11-17T18:10:00.000Z)\r\n18:01:30.906 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~> abc => 2\r\n18:01:30.906 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~~ [2016-11-17T18:00:00.000Z..2016-11-17T18:10:00.000Z)\r\n18:01:30.906 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~> abc => 2\r\n18:01:30.906 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~~ [2016-11-17T17:55:00.000Z..2016-11-17T18:05:00.000Z)\r\n18:01:30.906 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~> cab => 1\r\n18:01:30.906 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~~ [2016-11-17T17:55:00.000Z..2016-11-17T18:05:00.000Z)\r\n18:01:30.906 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~> cab => 1\r\n18:01:30.906 [main] INFO  c.q.m.m.core.mdef.SlidingWindowTest - ~~~~~~~~~~~ [2016-11-17T18:00:00.000Z..2016-11-17T18:10:00.000Z)\r\n...\r\n```\r\n\r\n**SDK version:** 1.8.0\r\n\r\nP/S: based on http://stackoverflow.com/questions/40527677/odd-behaviour-of-slidingwindows-when-used-with-testpipeline/40537541"},"comment":{"url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/issues/comments/261336881","html_url":"https://github.com/GoogleCloudPlatform/DataflowJavaSDK/issues/487#issuecomment-261336881","issue_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/issues/487","id":261336881,"user":{"login":"kennknowles","id":575353,"avatar_url":"https://avatars.githubusercontent.com/u/575353?v=3","gravatar_id":"","url":"https://api.github.com/users/kennknowles","html_url":"https://github.com/kennknowles","followers_url":"https://api.github.com/users/kennknowles/followers","following_url":"https://api.github.com/users/kennknowles/following{/other_user}","gists_url":"https://api.github.com/users/kennknowles/gists{/gist_id}","starred_url":"https://api.github.com/users/kennknowles/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/kennknowles/subscriptions","organizations_url":"https://api.github.com/users/kennknowles/orgs","repos_url":"https://api.github.com/users/kennknowles/repos","events_url":"https://api.github.com/users/kennknowles/events{/privacy}","received_events_url":"https://api.github.com/users/kennknowles/received_events","type":"User","site_admin":false},"created_at":"2016-11-17T19:01:32Z","updated_at":"2016-11-17T19:01:32Z","body":"Thanks for the detailed report! Taking a close look."}}
{"actor":{"display_login":"kennknowles"}}