Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/2090#discussion_r67009404
--- Diff:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamIterationTail.java
---
@@ -56,39 +57,60 @@ public void init() throws Exception {
LOG.info("Iteration tail {} acquired feedback queue {}",
getName(), brokerID);
- this.headOperator = new RecordPusher<>(dataChannel,
iterationWaitTime);
+ this.headOperator = new RecordPusher<>();
+ this.headOperator.setup(this, getConfiguration(), new
IterationTailOutput<>(dataChannel, iterationWaitTime));
}
private static class RecordPusher<IN> extends
AbstractStreamOperator<IN> implements OneInputStreamOperator<IN, IN> {
private static final long serialVersionUID = 1L;
+ @Override
+ public void processElement(StreamRecord<IN> record) throws
Exception {
+ output.collect(record);
+ }
+
+ @Override
+ public void processWatermark(Watermark mark) {
+ // ignore
+ }
+ }
+
+ private static class IterationTailOutput<IN> implements
Output<StreamRecord<IN>> {
+
@SuppressWarnings("NonSerializableFieldInSerializableClass")
private final BlockingQueue<StreamRecord<IN>> dataChannel;
private final long iterationWaitTime;
private final boolean shouldWait;
- RecordPusher(BlockingQueue<StreamRecord<IN>> dataChannel, long
iterationWaitTime) {
+ IterationTailOutput(BlockingQueue<StreamRecord<IN>>
dataChannel, long iterationWaitTime) {
this.dataChannel = dataChannel;
this.iterationWaitTime = iterationWaitTime;
this.shouldWait = iterationWaitTime > 0;
}
@Override
- public void processElement(StreamRecord<IN> record) throws
Exception {
- if (shouldWait) {
- dataChannel.offer(record, iterationWaitTime,
TimeUnit.MILLISECONDS);
- }
- else {
- dataChannel.put(record);
+ public void emitWatermark(Watermark mark) {
+ }
+
+ @Override
+ public void collect(StreamRecord<IN> record) {
+ try {
+ if (shouldWait) {
+ dataChannel.offer(record,
iterationWaitTime, TimeUnit.MILLISECONDS);
--- End diff --
Indention
---
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 [email protected] or file a JIRA ticket
with INFRA.
---
