You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2016/02/10 22:42:18 UTC
[1/2] flink git commit: [FLINK-3371] [api-breaking] Move
TriggerResult and TriggerContext to dedicated classes
Repository: flink
Updated Branches:
refs/heads/master 8df0bbacb -> fd324ea72
[FLINK-3371] [api-breaking] Move TriggerResult and TriggerContext to dedicated classes
This closes #1603
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/50bd65a5
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/50bd65a5
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/50bd65a5
Branch: refs/heads/master
Commit: 50bd65a574776817a03dd32fd438cb2327447109
Parents: 8df0bba
Author: Stephan Ewen <se...@apache.org>
Authored: Sun Feb 7 21:46:16 2016 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed Feb 10 22:15:31 2016 +0100
----------------------------------------------------------------------
.../examples/windowing/SessionWindowing.java | 3 +-
.../api/windowing/assigners/GlobalWindows.java | 10 +-
.../triggers/ContinuousEventTimeTrigger.java | 7 +-
.../ContinuousProcessingTimeTrigger.java | 2 +-
.../api/windowing/triggers/CountTrigger.java | 2 +-
.../api/windowing/triggers/DeltaTrigger.java | 7 +-
.../windowing/triggers/EventTimeTrigger.java | 5 +-
.../triggers/ProcessingTimeTrigger.java | 5 +-
.../api/windowing/triggers/PurgingTrigger.java | 4 +-
.../api/windowing/triggers/Trigger.java | 102 +++++--------------
.../api/windowing/triggers/TriggerResult.java | 96 +++++++++++++++++
.../windowing/EvictingWindowOperator.java | 5 +-
.../windowing/NonKeyedWindowOperator.java | 34 ++++---
.../operators/windowing/WindowOperator.java | 18 ++--
14 files changed, 179 insertions(+), 121 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/50bd65a5/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/SessionWindowing.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/SessionWindowing.java b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/SessionWindowing.java
index bd82800..e2df160 100644
--- a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/SessionWindowing.java
+++ b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/SessionWindowing.java
@@ -27,6 +27,7 @@ import org.apache.flink.streaming.api.functions.source.EventTimeSourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
+import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import java.util.ArrayList;
@@ -95,7 +96,7 @@ public class SessionWindowing {
env.execute();
}
- private static class SessionTrigger implements Trigger<Tuple3<String, Long, Integer>, GlobalWindow> {
+ private static class SessionTrigger extends Trigger<Tuple3<String, Long, Integer>, GlobalWindow> {
private static final long serialVersionUID = 1L;
http://git-wip-us.apache.org/repos/asf/flink/blob/50bd65a5/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/GlobalWindows.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/GlobalWindows.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/GlobalWindows.java
index d3eb2ac..a4d92cf 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/GlobalWindows.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/GlobalWindows.java
@@ -21,6 +21,7 @@ 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.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import java.util.Collection;
@@ -67,15 +68,12 @@ public class GlobalWindows extends WindowAssigner<Object, GlobalWindow> {
/**
* A trigger that never fires, as default Trigger for GlobalWindows.
*/
- private static class NeverTrigger implements Trigger<Object, GlobalWindow> {
+ private static class NeverTrigger extends Trigger<Object, GlobalWindow> {
private static final long serialVersionUID = 1L;
@Override
- public TriggerResult onElement(Object element,
- long timestamp,
- GlobalWindow window,
- TriggerContext ctx) {
- return TriggerResult.CONTINUE;
+ public TriggerResult onElement(Object element, long timestamp, GlobalWindow window, TriggerContext ctx) {
+ return TriggerResult.CONTINUE;
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/50bd65a5/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousEventTimeTrigger.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousEventTimeTrigger.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousEventTimeTrigger.java
index 02a935c..09f3959 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousEventTimeTrigger.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousEventTimeTrigger.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -7,7 +7,7 @@
* "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
+ * 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,
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.streaming.api.windowing.triggers;
import com.google.common.annotations.VisibleForTesting;
@@ -33,7 +34,7 @@ import org.apache.flink.streaming.api.windowing.windows.Window;
*
* @param <W> The type of {@link Window Windows} on which this trigger can operate.
*/
-public class ContinuousEventTimeTrigger<W extends Window> implements Trigger<Object, W> {
+public class ContinuousEventTimeTrigger<W extends Window> extends Trigger<Object, W> {
private static final long serialVersionUID = 1L;
private final long interval;
http://git-wip-us.apache.org/repos/asf/flink/blob/50bd65a5/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousProcessingTimeTrigger.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousProcessingTimeTrigger.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousProcessingTimeTrigger.java
index 25d9508..ca7ecb6 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousProcessingTimeTrigger.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousProcessingTimeTrigger.java
@@ -32,7 +32,7 @@ import org.apache.flink.streaming.api.windowing.windows.Window;
*
* @param <W> The type of {@link Window Windows} on which this trigger can operate.
*/
-public class ContinuousProcessingTimeTrigger<W extends Window> implements Trigger<Object, W> {
+public class ContinuousProcessingTimeTrigger<W extends Window> extends Trigger<Object, W> {
private static final long serialVersionUID = 1L;
private final long interval;
http://git-wip-us.apache.org/repos/asf/flink/blob/50bd65a5/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/CountTrigger.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/CountTrigger.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/CountTrigger.java
index 725cbf6..5113991 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/CountTrigger.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/CountTrigger.java
@@ -30,7 +30,7 @@ import java.io.IOException;
*
* @param <W> The type of {@link Window Windows} on which this trigger can operate.
*/
-public class CountTrigger<W extends Window> implements Trigger<Object, W> {
+public class CountTrigger<W extends Window> extends Trigger<Object, W> {
private static final long serialVersionUID = 1L;
private final long maxCount;
http://git-wip-us.apache.org/repos/asf/flink/blob/50bd65a5/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/DeltaTrigger.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/DeltaTrigger.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/DeltaTrigger.java
index 55c719a..4a6cde3 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/DeltaTrigger.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/DeltaTrigger.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -7,7 +7,7 @@
* "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
+ * 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,
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.streaming.api.windowing.triggers;
import org.apache.flink.api.common.state.ValueState;
@@ -33,7 +34,7 @@ import org.apache.flink.streaming.api.windowing.windows.Window;
*
* @param <W> The type of {@link Window Windows} on which this trigger can operate.
*/
-public class DeltaTrigger<T, W extends Window> implements Trigger<T, W> {
+public class DeltaTrigger<T, W extends Window> extends Trigger<T, W> {
private static final long serialVersionUID = 1L;
private final DeltaFunction<T> deltaFunction;
http://git-wip-us.apache.org/repos/asf/flink/blob/50bd65a5/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/EventTimeTrigger.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/EventTimeTrigger.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/EventTimeTrigger.java
index bbd0a01..17d04c2 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/EventTimeTrigger.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/EventTimeTrigger.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.streaming.api.windowing.triggers;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
@@ -25,7 +26,7 @@ import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
*
* @see org.apache.flink.streaming.api.watermark.Watermark
*/
-public class EventTimeTrigger implements Trigger<Object, TimeWindow> {
+public class EventTimeTrigger extends Trigger<Object, TimeWindow> {
private static final long serialVersionUID = 1L;
private EventTimeTrigger() {}
http://git-wip-us.apache.org/repos/asf/flink/blob/50bd65a5/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ProcessingTimeTrigger.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ProcessingTimeTrigger.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ProcessingTimeTrigger.java
index 85d6749..ae0b0e5 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ProcessingTimeTrigger.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ProcessingTimeTrigger.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.streaming.api.windowing.triggers;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
@@ -23,7 +24,7 @@ import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
* A {@link Trigger} that fires once the current system time passes the end of the window
* to which a pane belongs.
*/
-public class ProcessingTimeTrigger implements Trigger<Object, TimeWindow> {
+public class ProcessingTimeTrigger extends Trigger<Object, TimeWindow> {
private static final long serialVersionUID = 1L;
private ProcessingTimeTrigger() {}
http://git-wip-us.apache.org/repos/asf/flink/blob/50bd65a5/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/PurgingTrigger.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/PurgingTrigger.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/PurgingTrigger.java
index 626906c..0ec236b 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/PurgingTrigger.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/PurgingTrigger.java
@@ -25,12 +25,12 @@ import org.apache.flink.streaming.api.windowing.windows.Window;
*
* <p>
* When the nested trigger fires, this will return a {@code FIRE_AND_PURGE}
- * {@link org.apache.flink.streaming.api.windowing.triggers.Trigger.TriggerResult}
+ * {@link TriggerResult}
*
* @param <T> The type of elements on which this trigger can operate.
* @param <W> The type of {@link Window Windows} on which this trigger can operate.
*/
-public class PurgingTrigger<T, W extends Window> implements Trigger<T, W> {
+public class PurgingTrigger<T, W extends Window> extends Trigger<T, W> {
private static final long serialVersionUID = 1L;
private Trigger<T, W> nestedTrigger;
http://git-wip-us.apache.org/repos/asf/flink/blob/50bd65a5/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/Trigger.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/Trigger.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/Trigger.java
index fb61064..a200e5c 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/Trigger.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/Trigger.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.streaming.api.windowing.triggers;
import org.apache.flink.api.common.state.State;
@@ -39,12 +40,14 @@ import java.io.Serializable;
* <p>
* Triggers must not maintain state internally since they can be re-created or reused for
* different keys. All necessary state should be persisted using the state abstraction
- * available on the {@link org.apache.flink.streaming.api.windowing.triggers.Trigger.TriggerContext}.
+ * available on the {@link TriggerContext}.
*
* @param <T> The type of elements on which this {@code Trigger} works.
* @param <W> The type of {@link Window Windows} on which this {@code Trigger} can operate.
*/
-public interface Trigger<T, W extends Window> extends Serializable {
+public abstract class Trigger<T, W extends Window> implements Serializable {
+
+ private static final long serialVersionUID = -4104633972991191369L;
/**
* Called for every element that gets added to a pane. The result of this will determine
@@ -55,7 +58,7 @@ public interface Trigger<T, W extends Window> extends Serializable {
* @param window The window to which this pane belongs.
* @param ctx A context object that can be used to register timer callbacks.
*/
- TriggerResult onElement(T element, long timestamp, W window, TriggerContext ctx) throws Exception;
+ public abstract TriggerResult onElement(T element, long timestamp, W window, TriggerContext ctx) throws Exception;
/**
* Called when a processing-time timer that was set using the trigger context fires.
@@ -63,7 +66,7 @@ public interface Trigger<T, W extends Window> extends Serializable {
* @param time The timestamp at which the timer fired.
* @param ctx A context object that can be used to register timer callbacks.
*/
- TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception;
+ public abstract TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception;
/**
* Called when an event-time timer that was set using the trigger context fires.
@@ -71,102 +74,53 @@ public interface Trigger<T, W extends Window> extends Serializable {
* @param time The timestamp at which the timer fired.
* @param ctx A context object that can be used to register timer callbacks.
*/
- TriggerResult onEventTime(long time, W window, TriggerContext ctx) throws Exception;
+ public abstract TriggerResult onEventTime(long time, W window, TriggerContext ctx) throws Exception;
/**
* Clears any state that the trigger might still hold for the given window. This is called
* when a window is purged. Timers set using {@link TriggerContext#registerEventTimeTimer(long)}
* and {@link TriggerContext#registerProcessingTimeTimer(long)} should be deleted here as
* well as state acquired using {@link TriggerContext#getPartitionedState(StateDescriptor)}.
+ *
+ * <p>By default, this method does nothing.
*/
- void clear(W window, TriggerContext ctx) throws Exception;
+ public void clear(W window, TriggerContext ctx) throws Exception {}
+ // ------------------------------------------------------------------------
+
/**
- * Result type for trigger methods. This determines what happens with the window.
- *
- * <p>
- * On {@code FIRE} the pane is evaluated and results are emitted. The contents of the window
- * are kept. {@code FIRE_AND_PURGE} acts like {@code FIRE} but the contents of the pane
- * are purged. On {@code CONTINUE} nothing happens, processing continues. On {@code PURGE}
- * the contents of the window are discarded and no result is emitted for the window.
- */
- enum TriggerResult {
- CONTINUE(false, false), FIRE_AND_PURGE(true, true), FIRE(true, false), PURGE(false, true);
-
- private final boolean fire;
- private final boolean purge;
-
- TriggerResult(boolean fire, boolean purge) {
- this.purge = purge;
- this.fire = fire;
- }
-
- public boolean isFire() {
- return fire;
- }
-
- public boolean isPurge() {
- return purge;
- }
-
- /**
- * Merges two {@code TriggerResults}. This specifies what should happen if we have
- * two results from a Trigger, for example as a result from
- * {@link #onElement(Object, long, Window, TriggerContext)} and
- * {@link #onEventTime(long, Window, TriggerContext)}.
- *
- * <p>
- * For example, if one result says {@code CONTINUE} while the other says {@code FIRE}
- * then {@code FIRE} is the combined result;
- */
- public static TriggerResult merge(TriggerResult a, TriggerResult b) {
- if (a.purge || b.purge) {
- if (a.fire || b.fire) {
- return FIRE_AND_PURGE;
- } else {
- return PURGE;
- }
- } else if (a.fire || b.fire) {
- return FIRE;
- } else {
- return CONTINUE;
- }
- }
- }
-
- /**
- * A context object that is given to {@code Trigger} methods to allow them to register timer
+ * A context object that is given to {@link Trigger} methods to allow them to register timer
* callbacks and deal with state.
*/
- interface TriggerContext {
-
+ public interface TriggerContext {
+
/**
* Register a system time callback. When the current system time passes the specified
- * time {@link #onProcessingTime(long, Window, TriggerContext)} is called with the time specified here.
+ * time {@link Trigger#onProcessingTime(long, Window, TriggerContext)} is called with the time specified here.
*
- * @param time The time at which to invoke {@link #onProcessingTime(long, Window, TriggerContext)}
+ * @param time The time at which to invoke {@link Trigger#onProcessingTime(long, Window, TriggerContext)}
*/
void registerProcessingTimeTimer(long time);
-
+
/**
* Register an event-time callback. When the current watermark passes the specified
- * time {@link #onEventTime(long, Window, TriggerContext)} is called with the time specified here.
+ * time {@link Trigger#onEventTime(long, Window, TriggerContext)} is called with the time specified here.
*
- * @param time The watermark at which to invoke {@link #onEventTime(long, Window, TriggerContext)}
+ * @param time The watermark at which to invoke {@link Trigger#onEventTime(long, Window, TriggerContext)}
* @see org.apache.flink.streaming.api.watermark.Watermark
*/
void registerEventTimeTimer(long time);
-
+
/**
* Delete the processing time trigger for the given time.
*/
void deleteProcessingTimeTimer(long time);
-
+
/**
* Delete the event-time trigger for the given time.
*/
void deleteEventTimeTimer(long time);
-
+
/**
* Retrieves an {@link State} object that can be used to interact with
* fault-tolerant state that is scoped to the window and key of the current
@@ -180,7 +134,7 @@ public interface Trigger<T, W extends Window> extends Serializable {
* function (function is not part os a KeyedStream).
*/
<S extends State> S getPartitionedState(StateDescriptor<S, ?> stateDescriptor);
-
+
/**
* Retrieves a {@link ValueState} object that can be used to interact with
* fault-tolerant state that is scoped to the window and key of the current
@@ -199,8 +153,8 @@ public interface Trigger<T, W extends Window> extends Serializable {
*/
@Deprecated
<S extends Serializable> ValueState<S> getKeyValueState(String name, Class<S> stateType, S defaultState);
-
-
+
+
/**
* Retrieves a {@link ValueState} object that can be used to interact with
* fault-tolerant state that is scoped to the window and key of the current
http://git-wip-us.apache.org/repos/asf/flink/blob/50bd65a5/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/TriggerResult.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/TriggerResult.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/TriggerResult.java
new file mode 100644
index 0000000..2841542
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/TriggerResult.java
@@ -0,0 +1,96 @@
+/*
+ * 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.triggers;
+
+import org.apache.flink.streaming.api.windowing.windows.Window;
+
+/**
+ * Result type for trigger methods. This determines what happens with the window,
+ * for example whether the window function should be called, or the window
+ * should be discarded.
+ */
+public enum TriggerResult {
+
+ /**
+ * No action is taken on the window.
+ */
+ CONTINUE(false, false),
+
+ /**
+ * {@code FIRE_AND_PURGE} evaluates the window function and emits the window
+ * result.
+ */
+ FIRE_AND_PURGE(true, true),
+
+ /**
+ * On {@code FIRE}, the window is evaluated and results are emitted.
+ * The window is not purged, though, all elements are retained.
+ */
+ FIRE(true, false),
+
+ /**
+ * All elements in the window are cleared and the window is discarded,
+ * without evaluating the window function or emitting any elements.
+ */
+ PURGE(false, true);
+
+ // ------------------------------------------------------------------------
+
+ private final boolean fire;
+ private final boolean purge;
+
+ TriggerResult(boolean fire, boolean purge) {
+ this.purge = purge;
+ this.fire = fire;
+ }
+
+ public boolean isFire() {
+ return fire;
+ }
+
+ public boolean isPurge() {
+ return purge;
+ }
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * Merges two {@code TriggerResults}. This specifies what should happen if we have
+ * two results from a Trigger, for example as a result from
+ * {@link Trigger#onElement(Object, long, Window, Trigger.TriggerContext)} and
+ * {@link Trigger#onEventTime(long, Window, Trigger.TriggerContext)}.
+ *
+ * <p>
+ * For example, if one result says {@code CONTINUE} while the other says {@code FIRE}
+ * then {@code FIRE} is the combined result;
+ */
+ public static TriggerResult merge(TriggerResult a, TriggerResult b) {
+ if (a.purge || b.purge) {
+ if (a.fire || b.fire) {
+ return FIRE_AND_PURGE;
+ } else {
+ return PURGE;
+ }
+ } else if (a.fire || b.fire) {
+ return FIRE;
+ } else {
+ return CONTINUE;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/50bd65a5/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperator.java
index 41ec91a..a960ac4 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperator.java
@@ -30,6 +30,7 @@ import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.evictors.Evictor;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
+import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -87,7 +88,7 @@ public class EvictingWindowOperator<K, IN, OUT, W extends Window> extends Window
context.key = key;
context.window = window;
- Trigger.TriggerResult triggerResult = context.onElement(element);
+ TriggerResult triggerResult = context.onElement(element);
processTriggerResult(triggerResult, key, window);
}
@@ -95,7 +96,7 @@ public class EvictingWindowOperator<K, IN, OUT, W extends Window> extends Window
@Override
@SuppressWarnings("unchecked,rawtypes")
- protected void processTriggerResult(Trigger.TriggerResult triggerResult, K key, W window) throws Exception {
+ protected void processTriggerResult(TriggerResult triggerResult, K key, W window) throws Exception {
if (!triggerResult.isFire() && !triggerResult.isPurge()) {
// do nothing
return;
http://git-wip-us.apache.org/repos/asf/flink/blob/50bd65a5/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/NonKeyedWindowOperator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/NonKeyedWindowOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/NonKeyedWindowOperator.java
index d7dbaf5..93761e6 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/NonKeyedWindowOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/NonKeyedWindowOperator.java
@@ -38,6 +38,8 @@ import org.apache.flink.streaming.api.operators.TimestampedCollector;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
+import org.apache.flink.streaming.api.windowing.triggers.Trigger.TriggerContext;
+import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.streaming.runtime.operators.Triggerable;
import org.apache.flink.streaming.runtime.operators.windowing.buffers.WindowBuffer;
@@ -248,7 +250,7 @@ public class NonKeyedWindowOperator<IN, OUT, W extends Window>
windows.put(window, context);
}
context.windowBuffer.storeElement(element);
- Trigger.TriggerResult triggerResult = context.onElement(element);
+ TriggerResult triggerResult = context.onElement(element);
processTriggerResult(triggerResult, window);
}
}
@@ -264,7 +266,7 @@ public class NonKeyedWindowOperator<IN, OUT, W extends Window>
}
}
- private void processTriggerResult(Trigger.TriggerResult triggerResult, W window) throws Exception {
+ private void processTriggerResult(TriggerResult triggerResult, W window) throws Exception {
if (!triggerResult.isFire() && !triggerResult.isPurge()) {
// do nothing
return;
@@ -311,7 +313,7 @@ public class NonKeyedWindowOperator<IN, OUT, W extends Window>
// We have to check here whether the entry in the set still reflects the
// currently set timer in the Context.
if (ctx.watermarkTimer <= mark.getTimestamp()) {
- Trigger.TriggerResult triggerResult = ctx.onEventTime(ctx.watermarkTimer);
+ TriggerResult triggerResult = ctx.onEventTime(ctx.watermarkTimer);
processTriggerResult(triggerResult, ctx.window);
}
}
@@ -343,7 +345,7 @@ public class NonKeyedWindowOperator<IN, OUT, W extends Window>
// performance reasons. We have to check here whether the entry in the set still
// reflects the currently set timer in the Context.
if (ctx.processingTimeTimer <= time) {
- Trigger.TriggerResult triggerResult = ctx.onProcessingTime(ctx.processingTimeTimer);
+ TriggerResult triggerResult = ctx.onProcessingTime(ctx.processingTimeTimer);
processTriggerResult(triggerResult, ctx.window);
}
}
@@ -360,7 +362,7 @@ public class NonKeyedWindowOperator<IN, OUT, W extends Window>
* {@link org.apache.flink.streaming.api.windowing.assigners.WindowAssigner}. These panes all
* have their own instance of the {@code Trigger}.
*/
- protected class Context implements Trigger.TriggerContext {
+ protected class Context implements TriggerContext {
protected W window;
protected WindowBuffer<IN> windowBuffer;
@@ -538,41 +540,41 @@ public class NonKeyedWindowOperator<IN, OUT, W extends Window>
}
- public Trigger.TriggerResult onElement(StreamRecord<IN> element) throws Exception {
- Trigger.TriggerResult onElementResult = trigger.onElement(element.getValue(), element.getTimestamp(), window, this);
+ public TriggerResult onElement(StreamRecord<IN> element) throws Exception {
+ TriggerResult onElementResult = trigger.onElement(element.getValue(), element.getTimestamp(), window, this);
if (watermarkTimer > 0 && watermarkTimer <= currentWatermark) {
// fire now and don't wait for the next watermark update
- Trigger.TriggerResult onEventTimeResult = onEventTime(watermarkTimer);
- return Trigger.TriggerResult.merge(onElementResult, onEventTimeResult);
+ TriggerResult onEventTimeResult = onEventTime(watermarkTimer);
+ return TriggerResult.merge(onElementResult, onEventTimeResult);
} else {
return onElementResult;
}
}
- public Trigger.TriggerResult onProcessingTime(long time) throws Exception {
+ public TriggerResult onProcessingTime(long time) throws Exception {
if (time == processingTimeTimer) {
processingTimeTimer = -1;
return trigger.onProcessingTime(time, window, this);
} else {
- return Trigger.TriggerResult.CONTINUE;
+ return TriggerResult.CONTINUE;
}
}
- public Trigger.TriggerResult onEventTime(long time) throws Exception {
+ public TriggerResult onEventTime(long time) throws Exception {
if (time == watermarkTimer) {
watermarkTimer = -1;
- Trigger.TriggerResult firstTriggerResult = trigger.onEventTime(time, window, this);
+ TriggerResult firstTriggerResult = trigger.onEventTime(time, window, this);
if (watermarkTimer > 0 && watermarkTimer <= currentWatermark) {
// fire now and don't wait for the next watermark update
- Trigger.TriggerResult secondTriggerResult = onEventTime(watermarkTimer);
- return Trigger.TriggerResult.merge(firstTriggerResult, secondTriggerResult);
+ TriggerResult secondTriggerResult = onEventTime(watermarkTimer);
+ return TriggerResult.merge(firstTriggerResult, secondTriggerResult);
} else {
return firstTriggerResult;
}
} else {
- return Trigger.TriggerResult.CONTINUE;
+ return TriggerResult.CONTINUE;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/50bd65a5/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java
index eccaeee..5fc89e8 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java
@@ -40,6 +40,8 @@ import org.apache.flink.streaming.api.operators.TimestampedCollector;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
+import org.apache.flink.streaming.api.windowing.triggers.Trigger.TriggerContext;
+import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.streaming.runtime.operators.Triggerable;
import org.apache.flink.streaming.runtime.operators.windowing.buffers.WindowBufferFactory;
@@ -243,13 +245,13 @@ public class WindowOperator<K, IN, ACC, OUT, W extends Window>
context.key = key;
context.window = window;
- Trigger.TriggerResult triggerResult = context.onElement(element);
+ TriggerResult triggerResult = context.onElement(element);
processTriggerResult(triggerResult, key, window);
}
}
- protected void processTriggerResult(Trigger.TriggerResult triggerResult, K key, W window) throws Exception {
+ protected void processTriggerResult(TriggerResult triggerResult, K key, W window) throws Exception {
if (!triggerResult.isFire() && !triggerResult.isPurge()) {
// do nothing
return;
@@ -293,7 +295,7 @@ public class WindowOperator<K, IN, ACC, OUT, W extends Window>
context.key = timer.key;
context.window = timer.window;
setKeyContext(timer.key);
- Trigger.TriggerResult triggerResult = context.onEventTime(timer.timestamp);
+ TriggerResult triggerResult = context.onEventTime(timer.timestamp);
processTriggerResult(triggerResult, context.key, context.window);
} else {
fire = false;
@@ -320,7 +322,7 @@ public class WindowOperator<K, IN, ACC, OUT, W extends Window>
context.key = timer.key;
context.window = timer.window;
setKeyContext(timer.key);
- Trigger.TriggerResult triggerResult = context.onProcessingTime(timer.timestamp);
+ TriggerResult triggerResult = context.onProcessingTime(timer.timestamp);
processTriggerResult(triggerResult, context.key, context.window);
} else {
fire = false;
@@ -338,7 +340,7 @@ public class WindowOperator<K, IN, ACC, OUT, W extends Window>
* by setting the {@code key} and {@code window} fields. No internal state must be kept in
* the {@code Context}
*/
- protected class Context implements Trigger.TriggerContext {
+ protected class Context implements TriggerContext {
protected K key;
protected W window;
@@ -427,15 +429,15 @@ public class WindowOperator<K, IN, ACC, OUT, W extends Window>
}
- public Trigger.TriggerResult onElement(StreamRecord<IN> element) throws Exception {
+ public TriggerResult onElement(StreamRecord<IN> element) throws Exception {
return trigger.onElement(element.getValue(), element.getTimestamp(), window, this);
}
- public Trigger.TriggerResult onProcessingTime(long time) throws Exception {
+ public TriggerResult onProcessingTime(long time) throws Exception {
return trigger.onProcessingTime(time, window, this);
}
- public Trigger.TriggerResult onEventTime(long time) throws Exception {
+ public TriggerResult onEventTime(long time) throws Exception {
return trigger.onEventTime(time, window, this);
}
[2/2] flink git commit: [FLINK-3384] [kafka] Add ClosableQueue for
message exchanges between Kafka Threads
Posted by se...@apache.org.
[FLINK-3384] [kafka] Add ClosableQueue for message exchanges between Kafka Threads
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/fd324ea7
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/fd324ea7
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/fd324ea7
Branch: refs/heads/master
Commit: fd324ea72979cc3d4202ffa3ea174ec4cc9d153b
Parents: 50bd65a
Author: Stephan Ewen <se...@apache.org>
Authored: Wed Feb 10 14:51:10 2016 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed Feb 10 22:15:32 2016 +0100
----------------------------------------------------------------------
.../kafka/internals/ClosableBlockingQueue.java | 502 +++++++++++++++
.../internals/ClosableBlockingQueueTest.java | 603 +++++++++++++++++++
2 files changed, 1105 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/fd324ea7/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueue.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueue.java b/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueue.java
new file mode 100644
index 0000000..856c2ad
--- /dev/null
+++ b/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueue.java
@@ -0,0 +1,502 @@
+/*
+ * 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.connectors.kafka.internals;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * A special form of blocking queue with two additions:
+ * <ol>
+ * <li>The queue can be closed atomically when empty. Adding elements after the queue
+ * is closed fails. This allows queue consumers to atomically discover that no elements
+ * are available and mark themselves as shut down.</li>
+ * <li>The queue allows to poll batches of elements in one polling call.</li>
+ * </ol>
+ *
+ * The queue has no capacity restriction and is safe for multiple producers and consumers.
+ *
+ * <p>Note: Null elements are prohibited.
+ *
+ * @param <E> The type of elements in the queue.
+ */
+public class ClosableBlockingQueue<E> {
+
+ /** The lock used to make queue accesses and open checks atomic */
+ private final ReentrantLock lock;
+
+ /** The condition on which blocking get-calls wait if the queue is empty */
+ private final Condition nonEmpty;
+
+ /** The deque of elements */
+ private final ArrayDeque<E> elements;
+
+ /** Flag marking the status of the queue */
+ private volatile boolean open;
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * Creates a new empty queue.
+ */
+ public ClosableBlockingQueue() {
+ this(10);
+ }
+
+ /**
+ * Creates a new empty queue, reserving space for at least the specified number
+ * of elements. The queu can still grow, of more elements are added than the
+ * reserved space.
+ *
+ * @param initialSize The number of elements to reserve space for.
+ */
+ public ClosableBlockingQueue(int initialSize) {
+ this.lock = new ReentrantLock(true);
+ this.nonEmpty = this.lock.newCondition();
+
+ this.elements = new ArrayDeque<>(initialSize);
+ this.open = true;
+
+
+ }
+
+ /**
+ * Creates a new queue that contains the given elements.
+ *
+ * @param initialElements The elements to initially add to the queue.
+ */
+ public ClosableBlockingQueue(Collection<? extends E> initialElements) {
+ this(initialElements.size());
+ this.elements.addAll(initialElements);
+ }
+
+ // ------------------------------------------------------------------------
+ // Size and status
+ // ------------------------------------------------------------------------
+
+ /**
+ * Gets the number of elements currently in the queue.
+ * @return The number of elements currently in the queue.
+ */
+ public int size() {
+ return elements.size();
+ }
+
+ /**
+ * Checks whether the queue is empty (has no elements).
+ * @return True, if the queue is empty; false, if it is non-empty.
+ */
+ public boolean isEmpty() {
+ return size() == 0;
+ }
+
+ /**
+ * Checks whether the queue is currently open, meaning elements can be added and polled.
+ * @return True, if the queue is open; false, if it is closed.
+ */
+ public boolean isOpen() {
+ return open;
+ }
+
+ /**
+ * Tries to close the queue. Closing the queue only succeeds when no elements are
+ * in the queue when this method is called. Checking whether the queue is empty, and
+ * marking the queue as closed is one atomic operation.
+ *
+ * @return True, if the queue is closed, false if the queue remains open.
+ */
+ public boolean close() {
+ lock.lock();
+ try {
+ if (open) {
+ if (elements.isEmpty()) {
+ open = false;
+ nonEmpty.signalAll();
+ return true;
+ } else {
+ return false;
+ }
+ }
+ else {
+ // already closed
+ return true;
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Adding / Removing elements
+ // ------------------------------------------------------------------------
+
+ /**
+ * Tries to add an element to the queue, if the queue is still open. Checking whether the queue
+ * is open and adding the element is one atomic operation.
+ *
+ * <p>Unlike the {@link #add(Object)} method, this method never throws an exception,
+ * but only indicates via the return code if the element was added or the
+ * queue was closed.
+ *
+ * @param element The element to add.
+ * @return True, if the element was added, false if the queue was closes.
+ */
+ public boolean addIfOpen(E element) {
+ requireNonNull(element);
+
+ lock.lock();
+ try {
+ if (open) {
+ elements.addLast(element);
+ if (elements.size() == 1) {
+ nonEmpty.signalAll();
+ }
+ }
+ return open;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Adds the element to the queue, or fails with an exception, if the queue is closed.
+ * Checking whether the queue is open and adding the element is one atomic operation.
+ *
+ * @param element The element to add.
+ * @throws IllegalStateException Thrown, if the queue is closed.
+ */
+ public void add(E element) throws IllegalStateException {
+ requireNonNull(element);
+
+ lock.lock();
+ try {
+ if (open) {
+ elements.addLast(element);
+ if (elements.size() == 1) {
+ nonEmpty.signalAll();
+ }
+ } else {
+ throw new IllegalStateException("queue is closed");
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Returns the queue's next element without removing it, if the queue is non-empty.
+ * Otherwise, returns null.
+ *
+ * <p>The method throws an {@code IllegalStateException} if the queue is closed.
+ * Checking whether the queue is open and getting the next element is one atomic operation.
+ *
+ * <p>This method never blocks.
+ *
+ * @return The queue's next element, or null, if the queue is empty.
+ * @throws IllegalStateException Thrown, if the queue is closed.
+ */
+ public E peek() {
+ lock.lock();
+ try {
+ if (open) {
+ if (elements.size() > 0) {
+ return elements.getFirst();
+ } else {
+ return null;
+ }
+ } else {
+ throw new IllegalStateException("queue is closed");
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Returns the queue's next element and removes it, the queue is non-empty.
+ * Otherwise, this method returns null.
+ *
+ * <p>The method throws an {@code IllegalStateException} if the queue is closed.
+ * Checking whether the queue is open and removing the next element is one atomic operation.
+ *
+ * <p>This method never blocks.
+ *
+ * @return The queue's next element, or null, if the queue is empty.
+ * @throws IllegalStateException Thrown, if the queue is closed.
+ */
+ public E poll() {
+ lock.lock();
+ try {
+ if (open) {
+ if (elements.size() > 0) {
+ return elements.removeFirst();
+ } else {
+ return null;
+ }
+ } else {
+ throw new IllegalStateException("queue is closed");
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Returns all of the queue's current elements in a list, if the queue is non-empty.
+ * Otherwise, this method returns null.
+ *
+ * <p>The method throws an {@code IllegalStateException} if the queue is closed.
+ * Checking whether the queue is open and removing the elements is one atomic operation.
+ *
+ * <p>This method never blocks.
+ *
+ * @return All of the queue's elements, or null, if the queue is empty.
+ * @throws IllegalStateException Thrown, if the queue is closed.
+ */
+ public List<E> pollBatch() {
+ lock.lock();
+ try {
+ if (open) {
+ if (elements.size() > 0) {
+ ArrayList<E> result = new ArrayList<>(elements);
+ elements.clear();
+ return result;
+ } else {
+ return null;
+ }
+ } else {
+ throw new IllegalStateException("queue is closed");
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Returns the next element in the queue. If the queue is empty, this method
+ * waits until at least one element is added.
+ *
+ * <p>The method throws an {@code IllegalStateException} if the queue is closed.
+ * Checking whether the queue is open and removing the next element is one atomic operation.
+ *
+ * @return The next element in the queue, never null.
+ *
+ * @throws IllegalStateException Thrown, if the queue is closed.
+ * @throws InterruptedException Throw, if the thread is interrupted while waiting for an
+ * element to be added.
+ */
+ public E getElementBlocking() throws InterruptedException {
+ lock.lock();
+ try {
+ while (open && elements.isEmpty()) {
+ nonEmpty.await();
+ }
+
+ if (open) {
+ return elements.removeFirst();
+ } else {
+ throw new IllegalStateException("queue is closed");
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Returns the next element in the queue. If the queue is empty, this method
+ * waits at most a certain time until an element becomes available. If no element
+ * is available after that time, the method returns null.
+ *
+ * <p>The method throws an {@code IllegalStateException} if the queue is closed.
+ * Checking whether the queue is open and removing the next element is one atomic operation.
+ *
+ * @param timeoutMillis The number of milliseconds to block, at most.
+ * @return The next element in the queue, or null, if the timeout expires before an element is available.
+ *
+ * @throws IllegalStateException Thrown, if the queue is closed.
+ * @throws InterruptedException Throw, if the thread is interrupted while waiting for an
+ * element to be added.
+ */
+ public E getElementBlocking(long timeoutMillis) throws InterruptedException {
+ if (timeoutMillis == 0L) {
+ // wait forever case
+ return getElementBlocking();
+ } else if (timeoutMillis < 0L) {
+ throw new IllegalArgumentException("invalid timeout");
+ }
+
+ final long deadline = System.currentTimeMillis() + timeoutMillis;
+
+ lock.lock();
+ try {
+ while (open && elements.isEmpty() && timeoutMillis > 0) {
+ nonEmpty.await(timeoutMillis, TimeUnit.MILLISECONDS);
+ timeoutMillis = deadline - System.currentTimeMillis();
+ }
+
+ if (!open) {
+ throw new IllegalStateException("queue is closed");
+ }
+ else if (elements.isEmpty()) {
+ return null;
+ } else {
+ return elements.removeFirst();
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Gets all the elements found in the list, or blocks until at least one element
+ * was added. If the queue is empty when this method is called, it blocks until
+ * at least one element is added.
+ *
+ * <p>This method always returns a list with at least one element.
+ *
+ * <p>The method throws an {@code IllegalStateException} if the queue is closed.
+ * Checking whether the queue is open and removing the next element is one atomic operation.
+ *
+ * @return A list with all elements in the queue, always at least one element.
+ *
+ * @throws IllegalStateException Thrown, if the queue is closed.
+ * @throws InterruptedException Throw, if the thread is interrupted while waiting for an
+ * element to be added.
+ */
+ public List<E> getBatchBlocking() throws InterruptedException {
+ lock.lock();
+ try {
+ while (open && elements.isEmpty()) {
+ nonEmpty.await();
+ }
+ if (open) {
+ ArrayList<E> result = new ArrayList<>(elements);
+ elements.clear();
+ return result;
+ } else {
+ throw new IllegalStateException("queue is closed");
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Gets all the elements found in the list, or blocks until at least one element
+ * was added. This method is similar as {@link #getBatchBlocking()}, but takes
+ * a number of milliseconds that the method will maximally wait before returning.
+ *
+ * <p>This method never returns null, but an empty list, if the queue is empty when
+ * the method is called and the request times out before an element was added.
+ *
+ * <p>The method throws an {@code IllegalStateException} if the queue is closed.
+ * Checking whether the queue is open and removing the next element is one atomic operation.
+ *
+ * @param timeoutMillis The number of milliseconds to wait, at most.
+ * @return A list with all elements in the queue, possible an empty list.
+ *
+ * @throws IllegalStateException Thrown, if the queue is closed.
+ * @throws InterruptedException Throw, if the thread is interrupted while waiting for an
+ * element to be added.
+ */
+ public List<E> getBatchBlocking(long timeoutMillis) throws InterruptedException {
+ if (timeoutMillis == 0L) {
+ // wait forever case
+ return getBatchBlocking();
+ } else if (timeoutMillis < 0L) {
+ throw new IllegalArgumentException("invalid timeout");
+ }
+
+ final long deadline = System.currentTimeMillis() + timeoutMillis;
+
+ lock.lock();
+ try {
+ while (open && elements.isEmpty() && timeoutMillis > 0) {
+ nonEmpty.await(timeoutMillis, TimeUnit.MILLISECONDS);
+ timeoutMillis = deadline - System.currentTimeMillis();
+ }
+
+ if (!open) {
+ throw new IllegalStateException("queue is closed");
+ }
+ else if (elements.isEmpty()) {
+ return Collections.emptyList();
+ }
+ else {
+ ArrayList<E> result = new ArrayList<>(elements);
+ elements.clear();
+ return result;
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Standard Utilities
+ // ------------------------------------------------------------------------
+
+ @Override
+ public int hashCode() {
+ int hashCode = 17;
+ for (E element : elements) {
+ hashCode = 31 * hashCode + element.hashCode();
+ }
+ return hashCode;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == this) {
+ return true;
+ } else if (obj != null && obj.getClass() == ClosableBlockingQueue.class) {
+ @SuppressWarnings("unchecked")
+ ClosableBlockingQueue<E> that = (ClosableBlockingQueue<E>) obj;
+
+ if (this.elements.size() == that.elements.size()) {
+ Iterator<E> thisElements = this.elements.iterator();
+ for (E thatNext : that.elements) {
+ E thisNext = thisElements.next();
+ if (!(thisNext == null ? thatNext == null : thisNext.equals(thatNext))) {
+ return false;
+ }
+ }
+ return true;
+ } else {
+ return false;
+ }
+ } else {
+ return false;
+ }
+ }
+
+ @Override
+ public String toString() {
+ return elements.toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/fd324ea7/flink-streaming-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueueTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueueTest.java b/flink-streaming-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueueTest.java
new file mode 100644
index 0000000..6298c92
--- /dev/null
+++ b/flink-streaming-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueueTest.java
@@ -0,0 +1,603 @@
+/*
+ * 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.connectors.kafka.internals;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.*;
+import static java.util.Arrays.asList;
+import static java.util.Collections.emptyList;
+import static java.util.Collections.singletonList;
+
+public class ClosableBlockingQueueTest {
+
+ // ------------------------------------------------------------------------
+ // single-threaded unit tests
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testCreateQueueHashCodeEquals() {
+ try {
+ ClosableBlockingQueue<String> queue1 = new ClosableBlockingQueue<>();
+ ClosableBlockingQueue<String> queue2 = new ClosableBlockingQueue<>(22);
+
+ assertTrue(queue1.isOpen());
+ assertTrue(queue2.isOpen());
+ assertTrue(queue1.isEmpty());
+ assertTrue(queue2.isEmpty());
+ assertEquals(0, queue1.size());
+ assertEquals(0, queue2.size());
+
+ assertTrue(queue1.hashCode() == queue2.hashCode());
+ //noinspection EqualsWithItself
+ assertTrue(queue1.equals(queue1));
+ //noinspection EqualsWithItself
+ assertTrue(queue2.equals(queue2));
+ assertTrue(queue1.equals(queue2));
+
+ assertNotNull(queue1.toString());
+ assertNotNull(queue2.toString());
+
+ List<String> elements = new ArrayList<>();
+ elements.add("a");
+ elements.add("b");
+ elements.add("c");
+
+ ClosableBlockingQueue<String> queue3 = new ClosableBlockingQueue<>(elements);
+ ClosableBlockingQueue<String> queue4 = new ClosableBlockingQueue<>(asList("a", "b", "c"));
+
+ assertTrue(queue3.isOpen());
+ assertTrue(queue4.isOpen());
+ assertFalse(queue3.isEmpty());
+ assertFalse(queue4.isEmpty());
+ assertEquals(3, queue3.size());
+ assertEquals(3, queue4.size());
+
+ assertTrue(queue3.hashCode() == queue4.hashCode());
+ //noinspection EqualsWithItself
+ assertTrue(queue3.equals(queue3));
+ //noinspection EqualsWithItself
+ assertTrue(queue4.equals(queue4));
+ assertTrue(queue3.equals(queue4));
+
+ assertNotNull(queue3.toString());
+ assertNotNull(queue4.toString());
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testCloseEmptyQueue() {
+ try {
+ ClosableBlockingQueue<String> queue = new ClosableBlockingQueue<>();
+ assertTrue(queue.isOpen());
+ assertTrue(queue.close());
+ assertFalse(queue.isOpen());
+
+ assertFalse(queue.addIfOpen("element"));
+ assertTrue(queue.isEmpty());
+
+ try {
+ queue.add("some element");
+ fail("should cause an exception");
+ } catch (IllegalStateException ignored) {
+ // expected
+ }
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testCloseNonEmptyQueue() {
+ try {
+ ClosableBlockingQueue<Integer> queue = new ClosableBlockingQueue<>(asList(1, 2, 3));
+ assertTrue(queue.isOpen());
+
+ assertFalse(queue.close());
+ assertFalse(queue.close());
+
+ queue.poll();
+
+ assertFalse(queue.close());
+ assertFalse(queue.close());
+
+ queue.pollBatch();
+
+ assertTrue(queue.close());
+ assertFalse(queue.isOpen());
+
+ assertFalse(queue.addIfOpen(42));
+ assertTrue(queue.isEmpty());
+
+ try {
+ queue.add(99);
+ fail("should cause an exception");
+ } catch (IllegalStateException ignored) {
+ // expected
+ }
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testPeekAndPoll() {
+ try {
+ ClosableBlockingQueue<String> queue = new ClosableBlockingQueue<>();
+
+ assertNull(queue.peek());
+ assertNull(queue.peek());
+ assertNull(queue.poll());
+ assertNull(queue.poll());
+
+ assertEquals(0, queue.size());
+
+ queue.add("a");
+ queue.add("b");
+ queue.add("c");
+
+ assertEquals(3, queue.size());
+
+ assertEquals("a", queue.peek());
+ assertEquals("a", queue.peek());
+ assertEquals("a", queue.peek());
+
+ assertEquals(3, queue.size());
+
+ assertEquals("a", queue.poll());
+ assertEquals("b", queue.poll());
+
+ assertEquals(1, queue.size());
+
+ assertEquals("c", queue.peek());
+ assertEquals("c", queue.peek());
+
+ assertEquals("c", queue.poll());
+
+ assertEquals(0, queue.size());
+ assertNull(queue.poll());
+ assertNull(queue.peek());
+ assertNull(queue.peek());
+
+ assertTrue(queue.close());
+
+ try {
+ queue.peek();
+ fail("should cause an exception");
+ } catch (IllegalStateException ignored) {
+ // expected
+ }
+
+ try {
+ queue.poll();
+ fail("should cause an exception");
+ } catch (IllegalStateException ignored) {
+ // expected
+ }
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testPollBatch() {
+ try {
+ ClosableBlockingQueue<String> queue = new ClosableBlockingQueue<>();
+
+ assertNull(queue.pollBatch());
+
+ queue.add("a");
+ queue.add("b");
+
+ assertEquals(asList("a", "b"), queue.pollBatch());
+ assertNull(queue.pollBatch());
+
+ queue.add("c");
+
+ assertEquals(singletonList("c"), queue.pollBatch());
+ assertNull(queue.pollBatch());
+
+ assertTrue(queue.close());
+
+ try {
+ queue.pollBatch();
+ fail("should cause an exception");
+ } catch (IllegalStateException ignored) {
+ // expected
+ }
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testGetElementBlocking() {
+ try {
+ ClosableBlockingQueue<String> queue = new ClosableBlockingQueue<>();
+
+ assertNull(queue.getElementBlocking(1));
+ assertNull(queue.getElementBlocking(3));
+ assertNull(queue.getElementBlocking(2));
+
+ assertEquals(0, queue.size());
+
+ queue.add("a");
+ queue.add("b");
+ queue.add("c");
+ queue.add("d");
+ queue.add("e");
+ queue.add("f");
+
+ assertEquals(6, queue.size());
+
+ assertEquals("a", queue.getElementBlocking(99));
+ assertEquals("b", queue.getElementBlocking());
+
+ assertEquals(4, queue.size());
+
+ assertEquals("c", queue.getElementBlocking(0));
+ assertEquals("d", queue.getElementBlocking(1000000));
+ assertEquals("e", queue.getElementBlocking());
+ assertEquals("f", queue.getElementBlocking(1786598));
+
+ assertEquals(0, queue.size());
+
+ assertNull(queue.getElementBlocking(1));
+ assertNull(queue.getElementBlocking(3));
+ assertNull(queue.getElementBlocking(2));
+
+ assertTrue(queue.close());
+
+ try {
+ queue.getElementBlocking();
+ fail("should cause an exception");
+ } catch (IllegalStateException ignored) {
+ // expected
+ }
+
+ try {
+ queue.getElementBlocking(1000000000L);
+ fail("should cause an exception");
+ } catch (IllegalStateException ignored) {
+ // expected
+ }
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testGetBatchBlocking() {
+ try {
+ ClosableBlockingQueue<String> queue = new ClosableBlockingQueue<>();
+
+ assertEquals(emptyList(), queue.getBatchBlocking(1));
+ assertEquals(emptyList(), queue.getBatchBlocking(3));
+ assertEquals(emptyList(), queue.getBatchBlocking(2));
+
+ queue.add("a");
+ queue.add("b");
+
+ assertEquals(asList("a", "b"), queue.getBatchBlocking(900000009));
+
+ queue.add("c");
+ queue.add("d");
+
+ assertEquals(asList("c", "d"), queue.getBatchBlocking());
+
+ assertEquals(emptyList(), queue.getBatchBlocking(2));
+
+ queue.add("e");
+
+ assertEquals(singletonList("e"), queue.getBatchBlocking(0));
+
+ queue.add("f");
+
+ assertEquals(singletonList("f"), queue.getBatchBlocking(1000000000));
+
+ assertEquals(0, queue.size());
+
+ assertEquals(emptyList(), queue.getBatchBlocking(1));
+ assertEquals(emptyList(), queue.getBatchBlocking(3));
+ assertEquals(emptyList(), queue.getBatchBlocking(2));
+
+ assertTrue(queue.close());
+
+ try {
+ queue.getBatchBlocking();
+ fail("should cause an exception");
+ } catch (IllegalStateException ignored) {
+ // expected
+ }
+
+ try {
+ queue.getBatchBlocking(1000000000L);
+ fail("should cause an exception");
+ } catch (IllegalStateException ignored) {
+ // expected
+ }
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // multi-threaded tests
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void notifyOnClose() {
+ try {
+ final long oneYear = 365L * 24 * 60 * 60 * 1000;
+
+ // test "getBatchBlocking()"
+ final ClosableBlockingQueue<String> queue1 = new ClosableBlockingQueue<>();
+ QueueCall call1 = new QueueCall() {
+ @Override
+ public void call() throws Exception {
+ queue1.getBatchBlocking();
+ }
+ };
+ testCallExitsOnClose(call1, queue1);
+
+ // test "getBatchBlocking()"
+ final ClosableBlockingQueue<String> queue2 = new ClosableBlockingQueue<>();
+ QueueCall call2 = new QueueCall() {
+ @Override
+ public void call() throws Exception {
+ queue2.getBatchBlocking(oneYear);
+ }
+ };
+ testCallExitsOnClose(call2, queue2);
+
+ // test "getBatchBlocking()"
+ final ClosableBlockingQueue<String> queue3 = new ClosableBlockingQueue<>();
+ QueueCall call3 = new QueueCall() {
+ @Override
+ public void call() throws Exception {
+ queue3.getElementBlocking();
+ }
+ };
+ testCallExitsOnClose(call3, queue3);
+
+ // test "getBatchBlocking()"
+ final ClosableBlockingQueue<String> queue4 = new ClosableBlockingQueue<>();
+ QueueCall call4 = new QueueCall() {
+ @Override
+ public void call() throws Exception {
+ queue4.getElementBlocking(oneYear);
+ }
+ };
+ testCallExitsOnClose(call4, queue4);
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
+ @Test
+ public void testMultiThreadedAddGet() {
+ try {
+ final ClosableBlockingQueue<Integer> queue = new ClosableBlockingQueue<>();
+ final AtomicReference<Throwable> pushErrorRef = new AtomicReference<>();
+ final AtomicReference<Throwable> pollErrorRef = new AtomicReference<>();
+
+ final int numElements = 2000;
+
+ Thread pusher = new Thread("pusher") {
+
+ @Override
+ public void run() {
+ try {
+ final Random rnd = new Random();
+ for (int i = 0; i < numElements; i++) {
+ queue.add(i);
+
+ // sleep a bit, sometimes
+ int sleepTime = rnd.nextInt(3);
+ if (sleepTime > 1) {
+ Thread.sleep(sleepTime);
+ }
+ }
+
+ while (true) {
+ if (queue.close()) {
+ break;
+ } else {
+ Thread.sleep(5);
+ }
+ }
+ } catch (Throwable t) {
+ pushErrorRef.set(t);
+ }
+ }
+ };
+ pusher.start();
+
+ Thread poller = new Thread("poller") {
+
+ @SuppressWarnings("InfiniteLoopStatement")
+ @Override
+ public void run() {
+ try {
+ int count = 0;
+
+ try {
+ final Random rnd = new Random();
+ int nextExpected = 0;
+
+ while (true) {
+ int getMethod = count % 7;
+ switch (getMethod) {
+ case 0: {
+ Integer next = queue.getElementBlocking(1);
+ if (next != null) {
+ assertEquals(nextExpected, next.intValue());
+ nextExpected++;
+ count++;
+ }
+ break;
+ }
+ case 1: {
+ List<Integer> nextList = queue.getBatchBlocking();
+ for (Integer next : nextList) {
+ assertNotNull(next);
+ assertEquals(nextExpected, next.intValue());
+ nextExpected++;
+ count++;
+ }
+ break;
+ }
+ case 2: {
+ List<Integer> nextList = queue.getBatchBlocking(1);
+ if (nextList != null) {
+ for (Integer next : nextList) {
+ assertNotNull(next);
+ assertEquals(nextExpected, next.intValue());
+ nextExpected++;
+ count++;
+ }
+ }
+ break;
+ }
+ case 3: {
+ Integer next = queue.poll();
+ if (next != null) {
+ assertEquals(nextExpected, next.intValue());
+ nextExpected++;
+ count++;
+ }
+ break;
+ }
+ case 4: {
+ List<Integer> nextList = queue.pollBatch();
+ if (nextList != null) {
+ for (Integer next : nextList) {
+ assertNotNull(next);
+ assertEquals(nextExpected, next.intValue());
+ nextExpected++;
+ count++;
+ }
+ }
+ break;
+ }
+ default: {
+ Integer next = queue.getElementBlocking();
+ assertNotNull(next);
+ assertEquals(nextExpected, next.intValue());
+ nextExpected++;
+ count++;
+ }
+ }
+
+ // sleep a bit, sometimes
+ int sleepTime = rnd.nextInt(3);
+ if (sleepTime > 1) {
+ Thread.sleep(sleepTime);
+ }
+ }
+ } catch (IllegalStateException e) {
+ // we get this once the queue is closed
+ assertEquals(numElements, count);
+ }
+ } catch (Throwable t) {
+ pollErrorRef.set(t);
+ }
+ }
+ };
+ poller.start();
+
+ pusher.join();
+ poller.join();
+
+ if (pushErrorRef.get() != null) {
+ Throwable t = pushErrorRef.get();
+ t.printStackTrace();
+ fail("Error in pusher: " + t.getMessage());
+ }
+ if (pollErrorRef.get() != null) {
+ Throwable t = pollErrorRef.get();
+ t.printStackTrace();
+ fail("Error in poller: " + t.getMessage());
+ }
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Utils
+ // ------------------------------------------------------------------------
+
+ private static void testCallExitsOnClose(
+ final QueueCall call, ClosableBlockingQueue<String> queue) throws Exception {
+
+ final AtomicReference<Throwable> errorRef = new AtomicReference<>();
+
+ Runnable runnable = new Runnable() {
+ @Override
+ public void run() {
+ try {
+ call.call();
+ } catch (Throwable t) {
+ errorRef.set(t);
+ }
+ }
+ };
+
+ Thread thread = new Thread(runnable);
+ thread.start();
+ Thread.sleep(100);
+ queue.close();
+ thread.join();
+
+ @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
+ Throwable cause = errorRef.get();
+ assertTrue(cause instanceof IllegalStateException);
+ }
+
+ private interface QueueCall {
+ void call() throws Exception;
+ }
+}