[GitHub] flink pull request: [FLINK-3975] [docs] Override baseurl when serv...
GitHub user dyanarose opened a pull request: https://github.com/apache/flink/pull/2040 [FLINK-3975] [docs] Override baseurl when serving docs locally Updating build_docs.sh to serve the docs locally with the correct url. This change makes build_docs.sh run the equivalent serve command to build_docs.bat. removing now unnecessary config file. You can merge this pull request into a Git repository by running: $ git pull https://github.com/dyanarose/flink FLINK-3975 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/flink/pull/2040.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #2040 commit 96af1da4e4770f2e6ed18d8413a6ea106927606e Author: Dyana Rose Date: 2016-05-26T20:40:07Z [FLINK-3975] [docs] Override baseurl when serving docs locally --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request: [FLINK-3975] [docs] Override baseurl when serv...
Github user dyanarose commented on a diff in the pull request: https://github.com/apache/flink/pull/2040#discussion_r65137533 --- Diff: docs/build_docs.sh --- @@ -54,7 +54,7 @@ JEKYLL_CMD="build" while getopts ":p" opt; do case $opt in p) - JEKYLL_CMD="serve --config _config.yml,_local_preview_conf.yml --watch" + JEKYLL_CMD="serve --baseurl= --watch" --- End diff -- @uce It looks like the `baseurl` switch came in with 1.0 https://github.com/jekyll/jekyll/blob/0ee1a75d16265abc250d0747c5f93bf2ad2daa05/site/_docs/upgrading/0-to-2.md#baseurl. I think `serve` not respecting the multiple config files may be a bug/feature with Jekyll as the docs say that `serve` takes all the same switches as `build` does. https://jekyllrb.com/docs/configuration/#serve-command-options. I'm looking to prove that when I get the chance and look into it to raise an issue to see if the code or the docs need changing. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #2077: [FLINK-4026] Fix code, grammar, and link issues in...
GitHub user dyanarose opened a pull request: https://github.com/apache/flink/pull/2077 [FLINK-4026] Fix code, grammar, and link issues in the Streaming documentation * fixing grammar issues with he streaming API section of the documentation that make it hard to follow in places. * fixing an incorrect code example, and places of unnecessary parentheses on the Windows page * adding a missing link for Kineses Streams on the Connectors index page. * correcting the nav position of the Pre-defined Timestamp Extractors / Watermark Emitters page You can merge this pull request into a Git repository by running: $ git pull https://github.com/dyanarose/flink FLINK-4026 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/flink/pull/2077.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #2077 commit 9c561fb4825d202cefbe382df36abce7a6d825b3 Author: Dyana Rose Date: 2016-05-18T21:14:15Z [FLINK-4026] Fix code, grammar, and link issues in the Streaming documentation --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #5295: [FLINK-8384] [streaming] Session Window Assigner w...
GitHub user dyanarose opened a pull request: https://github.com/apache/flink/pull/5295 [FLINK-8384] [streaming] Session Window Assigner with Dynamic Gaps ## What is the purpose of the change This PR adds the ability for the Session Window assigners to to have dynamic inactivity gaps in addition to the existing static inactivity gaps. **Behaviour of dynamic gaps within existing sessions:** - scenario 1 - the new timeout is prior to the old timeout. The old timeout (the furthest in the future) is respected. - scenario 2 - the new timeout is after the old timeout. The new timeout is respected. - scenario 3 - a session is in flight, a new timeout is calculated, however no new events arrive for that session after the new timeout is calculated. This session will not have its timeout changed ## Brief change log **What's New** - SessionWindowTimeGapExtractor\ - Generic Interface with one extract method that returns the time gap - DynamicEventTimeSessionWindows\ - Generic event time session window - DynamicProcessingTimeSessionWindows\ - Generic processing time session window - TypedEventTimeTrigger\ - Generic event time trigger - TypedProcessingTimeTrigger\ - Generic processing time trigger - Tests for all the above ## Verifying this change This change added tests and can be verified as follows: - added tests for the typed triggers that duplicate the existing trigger tests to prove parity - added unit tests for the dynamic session window assigners that mimic the existing static session window assigner tests to prove parity in the static case - added tests to the WindowOperatorTest class to prove the behaviour of changing inactivity gaps ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): (no) - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no, though the two typed trigger classes are marked `@Public(Evolving)`) - The serializers: (no) - The runtime per-record code paths (performance sensitive): (no) - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (no) - The S3 file system connector: (no) ## Documentation - Does this pull request introduce a new feature? (yes) - If yes, how is the feature documented? (docs && JavaDocs) You can merge this pull request into a Git repository by running: $ git pull https://github.com/SaleCycle/flink dynamic-session-window-gaps Alternatively you can review and apply these changes as the patch at: https://github.com/apache/flink/pull/5295.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #5295 commit 399522a3e23a51ce1e860e5e09499ef98a7e340d Author: Dyana Rose Date: 2018-01-10T15:50:00Z [FLINK-8384] [streaming] Dynamic Gap Session Window Assigner ---
[GitHub] flink pull request #5295: [FLINK-8384] [streaming] Session Window Assigner w...
Github user dyanarose commented on a diff in the pull request:
https://github.com/apache/flink/pull/5295#discussion_r161807828
--- Diff:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/DynamicEventTimeSessionWindows.java
---
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.windowing.assigners;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.windowing.triggers.Trigger;
+import
org.apache.flink.streaming.api.windowing.triggers.TypedEventTimeTrigger;
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
+
+import java.util.Collection;
+import java.util.Collections;
+
+/**
+ * A {@link WindowAssigner} that windows elements into sessions based on
the timestamp of the
+ * elements. Windows cannot overlap.
+ *
+ * For example, in order to window into windows with a dynamic time gap:
+ * {@code
+ * DataStream> in = ...;
+ * KeyedStream> keyed = in.keyBy(...);
+ * WindowedStream, String, TimeWindows> windowed =
+ * keyed.window(DynamicEventTimeSessionWindows.withDynamicGap({@link
SessionWindowTimeGapExtractor }));
+ * }
+ *
+ * @param The type of the input elements
+ */
+public class DynamicEventTimeSessionWindows extends
MergingWindowAssigner {
+ private static final long serialVersionUID = 1L;
--- End diff --
The Typed requirement comes from the desire to allow the
SessionWindowTimeGapExtractor to accept a correctly typed element.
To do that the Assigner itself needs to be typed, which means that the
trigger needs to be typed and so on.
If the SessionWindowTimeGapExtractor extract method instead took `object`,
requiring that the implementer cast it, then the new Typed classes wouldn't be
necessary.
I don't find that to be the most user friendly interface though, when the
type information is available. But, yeah, I'm not happy with having to
implement these exact copy classes either...
---
[GitHub] flink pull request #5295: [FLINK-8384] [streaming] Session Window Assigner w...
Github user dyanarose commented on a diff in the pull request:
https://github.com/apache/flink/pull/5295#discussion_r161807850
--- Diff:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/DynamicProcessingTimeSessionWindows.java
---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.windowing.assigners;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.windowing.triggers.Trigger;
+import
org.apache.flink.streaming.api.windowing.triggers.TypedProcessingTimeTrigger;
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
+
+import java.util.Collection;
+import java.util.Collections;
+
+/**
+ * A {@link WindowAssigner} that windows elements into sessions based on
the current processing
+ * time. Windows cannot overlap.
+ *
+ * For example, in order to window into windows with a dynamic time gap:
+ * {@code
+ * DataStream> in = ...;
+ * KeyedStream> keyed = in.keyBy(...);
+ * WindowedStream, String, TimeWindows> windowed =
+ *
keyed.window(DynamicProcessingTimeSessionWindows.withDynamicGap({@link
SessionWindowTimeGapExtractor }));
+ * }
+ *
+ * @param The type of the input elements
+ */
+public class DynamicProcessingTimeSessionWindows extends
MergingWindowAssigner {
+ private static final long serialVersionUID = 1L;
--- End diff --
as above
---
[GitHub] flink pull request #5295: [FLINK-8384] [streaming] Session Window Assigner w...
Github user dyanarose commented on a diff in the pull request:
https://github.com/apache/flink/pull/5295#discussion_r161807869
--- Diff:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/DynamicProcessingTimeSessionWindows.java
---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.windowing.assigners;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.windowing.triggers.Trigger;
+import
org.apache.flink.streaming.api.windowing.triggers.TypedProcessingTimeTrigger;
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
+
+import java.util.Collection;
+import java.util.Collections;
+
+/**
+ * A {@link WindowAssigner} that windows elements into sessions based on
the current processing
+ * time. Windows cannot overlap.
+ *
+ * For example, in order to window into windows with a dynamic time gap:
+ * {@code
+ * DataStream> in = ...;
+ * KeyedStream> keyed = in.keyBy(...);
+ * WindowedStream, String, TimeWindows> windowed =
+ *
keyed.window(DynamicProcessingTimeSessionWindows.withDynamicGap({@link
SessionWindowTimeGapExtractor }));
+ * }
+ *
+ * @param The type of the input elements
+ */
+public class DynamicProcessingTimeSessionWindows extends
MergingWindowAssigner {
+ private static final long serialVersionUID = 1L;
+
+ protected SessionWindowTimeGapExtractor
sessionWindowTimeGapExtractor;
+
+ protected
DynamicProcessingTimeSessionWindows(SessionWindowTimeGapExtractor
sessionWindowTimeGapExtractor) {
+ this.sessionWindowTimeGapExtractor =
sessionWindowTimeGapExtractor;
+ }
+
+ @Override
+ public Collection assignWindows(T element, long timestamp,
WindowAssignerContext context) {
+ long currentProcessingTime = context.getCurrentProcessingTime();
+ long sessionTimeout =
sessionWindowTimeGapExtractor.extract(element, timestamp, context);
+ if (sessionTimeout <= 0) {
+ throw new IllegalArgumentException("Dynamic session
time gap must satisfy 0 < gap");
+ }
+ return Collections.singletonList(new
TimeWindow(currentProcessingTime, currentProcessingTime + sessionTimeout));
+ }
+
+ @Override
+ public Trigger
getDefaultTrigger(StreamExecutionEnvironment env) {
+ return TypedProcessingTimeTrigger.create();
--- End diff --
as above
---
[GitHub] flink pull request #5295: [FLINK-8384] [streaming] Session Window Assigner w...
Github user dyanarose commented on a diff in the pull request:
https://github.com/apache/flink/pull/5295#discussion_r161807837
--- Diff:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/DynamicEventTimeSessionWindows.java
---
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.windowing.assigners;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.windowing.triggers.Trigger;
+import
org.apache.flink.streaming.api.windowing.triggers.TypedEventTimeTrigger;
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
+
+import java.util.Collection;
+import java.util.Collections;
+
+/**
+ * A {@link WindowAssigner} that windows elements into sessions based on
the timestamp of the
+ * elements. Windows cannot overlap.
+ *
+ * For example, in order to window into windows with a dynamic time gap:
+ * {@code
+ * DataStream> in = ...;
+ * KeyedStream> keyed = in.keyBy(...);
+ * WindowedStream, String, TimeWindows> windowed =
+ * keyed.window(DynamicEventTimeSessionWindows.withDynamicGap({@link
SessionWindowTimeGapExtractor }));
+ * }
+ *
+ * @param The type of the input elements
+ */
+public class DynamicEventTimeSessionWindows extends
MergingWindowAssigner {
+ private static final long serialVersionUID = 1L;
+
+ protected SessionWindowTimeGapExtractor
sessionWindowTimeGapExtractor;
+
+ protected
DynamicEventTimeSessionWindows(SessionWindowTimeGapExtractor
sessionWindowTimeGapExtractor) {
+ this.sessionWindowTimeGapExtractor =
sessionWindowTimeGapExtractor;
+ }
+
+ @Override
+ public Collection assignWindows(T element, long timestamp,
WindowAssignerContext context) {
+ long sessionTimeout =
sessionWindowTimeGapExtractor.extract(element, timestamp, context);
+ if (sessionTimeout <= 0) {
+ throw new IllegalArgumentException("Dynamic session
time gap must satisfy 0 < gap");
+ }
+ return Collections.singletonList(new TimeWindow(timestamp,
timestamp + sessionTimeout));
+ }
+
+ @Override
+ public Trigger
getDefaultTrigger(StreamExecutionEnvironment env) {
+ return TypedEventTimeTrigger.create();
--- End diff --
as above
---
[GitHub] flink pull request #5295: [FLINK-8384] [streaming] Session Window Assigner w...
Github user dyanarose commented on a diff in the pull request:
https://github.com/apache/flink/pull/5295#discussion_r161807900
--- Diff:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/SessionWindowTimeGapExtractor.java
---
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.windowing.assigners;
+
+import java.io.Serializable;
+
+/**
+ * A {@code SessionWindowTimeGapExtractor} extracts session time gaps for
Dynamic Session Window Assigners.
+ *
+ * @param The type of elements that this {@code
SessionWindowTimeGapExtractor} can extract session time gaps from.
+ */
+public interface SessionWindowTimeGapExtractor extends Serializable {
+ long extract(T element, long timestamp,
WindowAssigner.WindowAssignerContext context);
--- End diff --
can do.
Do you think it should return a Time instead of a long? To prevent anyone
trying to return seconds/minutes/whatever instead of milliseconds?
---
[GitHub] flink issue #5295: [FLINK-8384] [streaming] Session Window Assigner with Dyn...
Github user dyanarose commented on the issue: https://github.com/apache/flink/pull/5295 looks like the build failed on: org.apache.flink.test.streaming.runtime.StreamTaskTimerITCase testOperatorChainedToSource I can't see why this change would cause that to fail, after the PR passed originally. However I can't see it on the flaky test list ---
[GitHub] flink pull request #5295: [FLINK-8384] [streaming] Session Window Assigner w...
Github user dyanarose commented on a diff in the pull request:
https://github.com/apache/flink/pull/5295#discussion_r163692223
--- Diff:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/DynamicEventTimeSessionWindows.java
---
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.windowing.assigners;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.windowing.triggers.Trigger;
+import
org.apache.flink.streaming.api.windowing.triggers.TypedEventTimeTrigger;
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
+
+import java.util.Collection;
+import java.util.Collections;
+
+/**
+ * A {@link WindowAssigner} that windows elements into sessions based on
the timestamp of the
+ * elements. Windows cannot overlap.
+ *
+ * For example, in order to window into windows with a dynamic time gap:
+ * {@code
+ * DataStream> in = ...;
+ * KeyedStream> keyed = in.keyBy(...);
+ * WindowedStream, String, TimeWindows> windowed =
+ * keyed.window(DynamicEventTimeSessionWindows.withDynamicGap({@link
SessionWindowTimeGapExtractor }));
+ * }
+ *
+ * @param The type of the input elements
+ */
+public class DynamicEventTimeSessionWindows extends
MergingWindowAssigner {
+ private static final long serialVersionUID = 1L;
--- End diff --
changing EventTimeTrigger/ProcessingTimeTrigger would be my preference.
However I was hesitant to just go in and change the class definition. I also
don't believe it *should* break existing tests
I'll give it a go and see
---
[GitHub] flink pull request #5295: [FLINK-8384] [streaming] Session Window Assigner w...
Github user dyanarose commented on a diff in the pull request:
https://github.com/apache/flink/pull/5295#discussion_r163796598
--- Diff:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/DynamicEventTimeSessionWindows.java
---
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.windowing.assigners;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.windowing.triggers.Trigger;
+import
org.apache.flink.streaming.api.windowing.triggers.TypedEventTimeTrigger;
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
+
+import java.util.Collection;
+import java.util.Collections;
+
+/**
+ * A {@link WindowAssigner} that windows elements into sessions based on
the timestamp of the
+ * elements. Windows cannot overlap.
+ *
+ * For example, in order to window into windows with a dynamic time gap:
+ * {@code
+ * DataStream> in = ...;
+ * KeyedStream> keyed = in.keyBy(...);
+ * WindowedStream, String, TimeWindows> windowed =
+ * keyed.window(DynamicEventTimeSessionWindows.withDynamicGap({@link
SessionWindowTimeGapExtractor }));
+ * }
+ *
+ * @param The type of the input elements
+ */
+public class DynamicEventTimeSessionWindows extends
MergingWindowAssigner {
+ private static final long serialVersionUID = 1L;
--- End diff --
three sets of flink-streaming-scala tests needed to be updated to add
existential type information due to adding the generic to the EventTimeTrigger
and ProcessingTimeTrigger classes
---
[GitHub] flink issue #5295: [FLINK-8384] [streaming] Session Window Assigner with Dyn...
Github user dyanarose commented on the issue: https://github.com/apache/flink/pull/5295 I'm not the biggest fan of unchecked casts, but testing this in our POC environment casting the existing EventTimeTrigger to a typed Trigger is working. So if the unchecked cast is acceptable, that would get rid of all changes required to Event/ProcessingTimeTrigger On your second point. (sorry if I mis-attribute this based on github profile name) I believe you had mentioned breaking this out into new classes when I first brought this up on the mailing list https://lists.apache.org/thread.html/6ceb094460bca8e9731e9e1dc0bb479f407f769458bff30c412adf78@%3Cdev.flink.apache.org%3E Now that you see it in action, do you feel it would be better off as an addition to the existing session window classes? The way I see it is, if I put withGapExtractor() on the existing classes, without adding type information to them, then the extract() method on SessionWindowGapExtractor will need to have the signature of extract(Object input) leaving the implementer to have to cast to the input type. I have to admit it feels strange to me that the Window assigners all drop input type information. But that would mean that these new typed assigners would be the odd ones out. ---
[GitHub] flink issue #5295: [FLINK-8384] [streaming] Session Window Assigner with Dyn...
Github user dyanarose commented on the issue: https://github.com/apache/flink/pull/5295 Ah, I hadn't thought to keep both in place. So unless the Dynamic SessionWindow classes had withDynamicGap made package private, you would then be able to instantiate them from two different classes. That could feel a bit iffy, however someone else would call it a convenience method. I'll get the change in for the Trigger cast, that should clean up the PR a fair bit ---
[GitHub] flink issue #5295: [FLINK-8384] [streaming] Session Window Assigner with Dyn...
Github user dyanarose commented on the issue: https://github.com/apache/flink/pull/5295 the ci fail looks to be a known flaky test: FlinkKafkaProducer011ITCase.testScaleDownBeforeFirstCheckpoint ---
[GitHub] flink issue #5295: [FLINK-8384] [streaming] Session Window Assigner with Dyn...
Github user dyanarose commented on the issue: https://github.com/apache/flink/pull/5295 I like long myself, but I think that's only because I'm quite used to working in milliseconds. As the existing static Session Windows take Time as the gap, I think it made sense to have the extract method also produce a time. If it returns a Time, we don't have to worry about an implementer getting confused about what time unit they need to be returning, or always having to look it up just to check that they're right. ---
[GitHub] flink issue #5295: [FLINK-8384] [streaming] Session Window Assigner with Dyn...
Github user dyanarose commented on the issue: https://github.com/apache/flink/pull/5295 erf, I see what you mean, as well as the creation of all those Time objects. ---
[GitHub] flink issue #5295: [FLINK-8384] [streaming] Session Window Assigner with Dyn...
Github user dyanarose commented on the issue: https://github.com/apache/flink/pull/5295 the change to return Time has been backed out, so extract returns a long again. PublicEvolving annotations have been added to the new classes and methods. ---
[GitHub] flink issue #5295: [FLINK-8384] [streaming] Session Window Assigner with Dyn...
Github user dyanarose commented on the issue: https://github.com/apache/flink/pull/5295 I can see it's gone through Travis and is now in master, so closing as requested ---
[GitHub] flink pull request #5295: [FLINK-8384] [streaming] Session Window Assigner w...
Github user dyanarose closed the pull request at: https://github.com/apache/flink/pull/5295 ---
