[GitHub] flink issue #5295: [FLINK-8384] [streaming] Session Window Assigner with Dyn...

[dyanarose]

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 issue #5295: [FLINK-8384] [streaming] Session Window Assigner with Dyn...

[aljoscha]

Github user aljoscha commented on the issue:

https://github.com/apache/flink/pull/5295
  
Thanks a lot for working on this and iterating so quickly! 👍 

I merged this but could you please close the PR if it doesn't close 
automatically?


---

[GitHub] flink issue #5295: [FLINK-8384] [streaming] Session Window Assigner with Dyn...

[dyanarose]

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...

[dyanarose]

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...

[aljoscha]

Github user aljoscha commented on the issue:

https://github.com/apache/flink/pull/5295
  
I get that logic, but the existing `TimestampAssigner` also returns a 
`long` and if we return `Time` we always have to wrap/unwrap that long. What do 
you think?


---

[GitHub] flink issue #5295: [FLINK-8384] [streaming] Session Window Assigner with Dyn...

[dyanarose]

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...

[aljoscha]

Github user aljoscha commented on the issue:

https://github.com/apache/flink/pull/5295
  
I think the changes are good! Thanks for working on this. 👍 

As a final change before merging, I would annotate the new classes/methods 
as `@PublicEvolving`, would you be ok with that? And I would also like to 
change `SessionWindowTimeGapExtractor.extract()` to return a long instead of 
`Time`. What do you think?


---

[GitHub] flink issue #5295: [FLINK-8384] [streaming] Session Window Assigner with Dyn...

[dyanarose]

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...

[dyanarose]

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...

[aljoscha]

Github user aljoscha commented on the issue:

https://github.com/apache/flink/pull/5295
  
Yes, that was me. 😅 It was just a quick idea, and it would work because 
both `withGap()` and `withGapExtractor()` are static so the latter could have 
`T` on the method signature and return a `DynamicEventTimeSessionWindows` (or 
some such). I'm not against keeping it in the separate class, though.

I agree that the cast is a bit wonky but we know that it always works 
because the trigger we return never looks at the element.


---

[GitHub] flink issue #5295: [FLINK-8384] [streaming] Session Window Assigner with Dyn...

[dyanarose]

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...

[aljoscha]

Github user aljoscha commented on the issue:

https://github.com/apache/flink/pull/5295
  
I like the functionality of this a lot! However, I don't like that we 
change the signature of existing triggers or that we introduce new triggers 
that duplicate existing code.

As an alternative, could you cast the `EventTimeTrigger` to `Trigger` 
in `getDefaultTrigger()` of your new assigner?

Also an additional idea, instead of putting the API method on 
`DynamicEventTimeSessionWindows` we could think about adding it to 
`EventTimeSessionWindows`. We would then have 
`EventTimeSessionWindows.withGap()` and 
`EventTimeSessionWindows.withGapExtractor()`. What do you think?


---

[GitHub] flink issue #5295: [FLINK-8384] [streaming] Session Window Assigner with Dyn...

[sunjincheng121]

Github user sunjincheng121 commented on the issue:

https://github.com/apache/flink/pull/5295
  
It's `NoResourceAvailableException` error, not sure, but we can try to 
rebuild it. 


---

[GitHub] flink issue #5295: [FLINK-8384] [streaming] Session Window Assigner with Dyn...

[dyanarose]

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


---