[BEAM-3905] Update Flink Runner to Flink 1.5.0 #5578
Conversation
|
Run Flink ValidatesRunner |
|
|
||
| if (keySelector != null) { | ||
| ListStateDescriptor<WindowedValue<InputT>> pushedBackStateDescriptor = | ||
| new ListStateDescriptor<>( |
There was a problem hiding this comment.
Maybe move above of if/else given that is the same in both conditions.
| Stream<WindowedValue<InputT>> pushedBack = pushedBackElementsHandler.getElements(); | ||
| Iterator<WindowedValue<InputT>> it = pushedBack.iterator(); | ||
|
|
||
| long min = Long.MAX_VALUE; |
There was a problem hiding this comment.
Move out of the if condition to avoid the else, and then you do the setPushedBackWatermark on min for all cases.
There was a problem hiding this comment.
I'm not sure I can because I'm iterating over the pushed back elements, which I only have in the case where I have side inputs.
| if (keyedStateInternals == null) { | ||
| throw new RuntimeException("Current watermark is still " + currentOutputWatermark + "."); | ||
|
|
||
| } else { |
There was a problem hiding this comment.
Let just the if and this is the default.
There was a problem hiding this comment.
You mean flip the if?
| Iterator<WindowedValue<InputT>> it = pushedBack.iterator(); | ||
|
|
||
| long min = Long.MAX_VALUE; | ||
| while (it.hasNext()) { |
There was a problem hiding this comment.
Maybe 'simpler' to do min = pushedBack.map(v -> v.getTimestamp().getMillis()).reduce(Long.MAX_VALUE, (t1, t2) -> Math.min(t1, t2));
| @@ -418,33 +451,7 @@ public void close() throws Exception { | |||
| } | |||
|
|
|||
| private long getPushbackWatermarkHold() { | |||
There was a problem hiding this comment.
Maybe remove the method ? since now the state is kept in the variable, just access directly.
There was a problem hiding this comment.
I kept them for now to not have too many changes but can do that as a follow up
| @Override | ||
| public void pushBackAll(Iterable<T> elements) throws Exception { | ||
| for (T e : elements) { | ||
| // TODO: use addAll() once Flink 1.5.0 has addAll(Iterable<T>) |
There was a problem hiding this comment.
maybe just once Flink has addAll(Iterable) because Flink 1.5.0 does not have it.
| */ | ||
| class KeyedPushedBackElementsHandler<K, T> implements PushedBackElementsHandler<T> { | ||
|
|
||
| static <K, T> PushedBackElementsHandler<T> usingState( |
There was a problem hiding this comment.
create and return the concrete type KeyedPushedBackElementsHandler?
| if (lastSeenTimestamp == null) { | ||
| // we have never seen this, emit | ||
| WindowedValue<ValueWithRecordId<T>> value = streamRecord.getValue(); | ||
| output.collect(streamRecord.replace(value.withValue(value.getValue().getValue()))); |
There was a problem hiding this comment.
value.withValue(value.getValue().getValue() obvious :P
|
|
||
| Long lastSeenTimestamp = dedupingState.value(); | ||
| if (lastSeenTimestamp != null | ||
| && lastSeenTimestamp.equals(internalTimer.getTimestamp() - MAX_RETENTION_SINCE_ACCESS)) { |
There was a problem hiding this comment.
Question: Shouldn't this be like > or something like that or it has to be exact?
There was a problem hiding this comment.
I think it's exact because we really want to compare to the timer we set. But I think I just moved this code around.
| @@ -148,7 +148,9 @@ public void testRestore() throws Exception { | |||
| emptyMap(), | |||
| emptyList(), | |||
| PipelineOptionsFactory.as(FlinkPipelineOptions.class), | |||
| VarLongCoder.of() | |||
| VarLongCoder.of(), | |||
| null /* key selector */ // TODO use proper key selector | |||
There was a problem hiding this comment.
Maybe create a JIRA for this one if you don't plan to address it in this PR.
There was a problem hiding this comment.
apperently we don't need a key selector here so I'll remove the TODO
While doing this, we also have to re-do how we store non-keyed state in the DoFnOperator and fix usage of some interfaces that were removed.
aljoscha
force-pushed
the
jira-3905-update-to-flink-1.5
branch
from
190808e
to
c1958ce
Compare
|
I pushed a new version. Unfortunately, I messed up and don't have the fixups as a separate commit. 😓 |
While doing this, we also have to re-do how we store non-keyed state in
the DoFnOperator and fix usage of some interfaces that were removed.