You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by rm...@apache.org on 2015/02/02 19:42:12 UTC
[34/52] [partial] flink git commit: [FLINK-1452] Rename
'flink-addons' to 'flink-staging'
http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/EvictionPolicy.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/EvictionPolicy.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/EvictionPolicy.java
deleted file mode 100644
index c224ad4..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/EvictionPolicy.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * 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.policy;
-
-import java.io.Serializable;
-
-/**
- * An eviction policy specifies under which condition data points should be
- * deleted from the buffer. Deletions must be done only in the order the
- * elements arrived. Therefore, the policy only returns the number of elements
- * to evict on each element arrival.
- *
- * @param <DATA>
- * the type of the data handled by this policy
- */
-public interface EvictionPolicy<DATA> extends Serializable {
-
- /**
- * Proves if and how many elements should be deleted from the element
- * buffer. The eviction takes place after the trigger and after the call to
- * the UDF but before the adding of the new data point.
- *
- * @param datapoint
- * data point the data point which arrived
- * @param triggered
- * Information whether the UDF was triggered or not
- * @param bufferSize
- * The current size of the element buffer at the invokable
- * @return The number of elements to be deleted from the buffer
- */
- public int notifyEviction(DATA datapoint, boolean triggered, int bufferSize);
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/PunctuationPolicy.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/PunctuationPolicy.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/PunctuationPolicy.java
deleted file mode 100644
index ee8a3cb..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/PunctuationPolicy.java
+++ /dev/null
@@ -1,116 +0,0 @@
-/*
- * 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.policy;
-
-import org.apache.flink.streaming.api.windowing.extractor.Extractor;
-
-/**
- * This policy can be used to trigger and evict based on a punctuation which is
- * present within the arriving data. Using this policy, one can react on an
- * externally defined arbitrary windowing semantic.
- *
- * In case this policy is used for eviction, the complete buffer will get
- * deleted in case the punctuation is detected.
- *
- * By default this policy does not react on fake elements. Wrap it in an
- * {@link ActiveEvictionPolicyWrapper} to make it react on punctuation even in
- * fake elements.
- *
- * @param <IN>
- * The type of the input data handled by this policy. An
- * {@link Extractor} can be used to extract DATA for IN.
- * @param <DATA>
- * The type of the punctuation. An {@link Extractor} can be used to
- * extract DATA for IN.
- */
-public class PunctuationPolicy<IN, DATA> implements CloneableTriggerPolicy<IN>,
- CloneableEvictionPolicy<IN> {
-
- /**
- * auto generated version id
- */
- private static final long serialVersionUID = -8845130188912602498L;
- private int counter = 0;
- private Extractor<IN, DATA> extractor;
- private DATA punctuation;
-
- /**
- * Creates the punctuation policy without using any extractor. To make this
- * work IN and DATA must not be different types.
- *
- * @param punctuation
- * the punctuation which leads to trigger/evict.
- */
- public PunctuationPolicy(DATA punctuation) {
- this(punctuation, null);
- }
-
- /**
- * Creates the punctuation policy which uses the specified extractor to
- * isolate the punctuation from the data.
- *
- * @param punctuation
- * the punctuation which leads to trigger/evict.
- * @param extractor
- * An {@link Extractor} which converts IN to DATA.
- */
- public PunctuationPolicy(DATA punctuation, Extractor<IN, DATA> extractor) {
- this.punctuation = punctuation;
- this.extractor = extractor;
- }
-
- @Override
- public int notifyEviction(IN datapoint, boolean triggered, int bufferSize) {
- if (notifyTrigger(datapoint)) {
- int tmp = counter;
- // As the current will be add after the eviction the counter needs
- // to be set to one already
- counter = 1;
- return tmp;
- } else {
- counter++;
- return 0;
- }
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public boolean notifyTrigger(IN datapoint) {
- DATA tmp;
-
- // eventually extract data
- if (extractor == null) {
- // unchecked convert (cannot check it here)
- tmp = (DATA) datapoint;
- } else {
- tmp = extractor.extract(datapoint);
- }
-
- // compare data with punctuation
- if (punctuation.equals(tmp)) {
- return true;
- } else {
- return false;
- }
- }
-
- @Override
- public PunctuationPolicy<IN, DATA> clone() {
- return new PunctuationPolicy<IN, DATA>(punctuation, extractor);
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicy.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicy.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicy.java
deleted file mode 100644
index 16c30fc..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicy.java
+++ /dev/null
@@ -1,137 +0,0 @@
-/*
- * 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.policy;
-
-import java.util.LinkedList;
-
-import org.apache.flink.streaming.api.windowing.helper.Timestamp;
-import org.apache.flink.streaming.api.windowing.helper.TimestampWrapper;
-
-/**
- * This eviction policy evicts all elements which are older then a specified
- * time. The time is measured using a given {@link Timestamp} implementation. A
- * point in time is always represented as long. Therefore, the granularity can
- * be set as long value as well.
- *
- * @param <DATA>
- * The type of the incoming data points which are processed by this
- * policy.
- */
-public class TimeEvictionPolicy<DATA> implements ActiveEvictionPolicy<DATA>,
- CloneableEvictionPolicy<DATA> {
-
- /**
- * auto generated version id
- */
- private static final long serialVersionUID = -1457476766124518220L;
-
- private long granularity;
- private TimestampWrapper<DATA> timestampWrapper;
- private LinkedList<Long> buffer = new LinkedList<Long>();
-
- /**
- * This eviction policy evicts all elements which are older than a specified
- * time. The time is measured using a given {@link Timestamp}
- * implementation. A point in time is always represented as long. Therefore,
- * the granularity can be set as long value as well. If this value is set to
- * 2 the policy will evict all elements which are older as 2.
- *
- * <code>
- * while (time(firstInBuffer)<current time-granularity){
- * evict firstInBuffer;
- * }
- * </code>
- *
- * @param granularity
- * The granularity of the eviction. If this value is set to 2 the
- * policy will evict all elements which are older as 2(if
- * (time(X)<current time-granularity) evict X).
- * @param timestampWrapper
- * The {@link TimestampWrapper} to measure the time with. This
- * can be either user defined of provided by the API.
- */
- public TimeEvictionPolicy(long granularity, TimestampWrapper<DATA> timestampWrapper) {
- this.timestampWrapper = timestampWrapper;
- this.granularity = granularity;
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public int notifyEvictionWithFakeElement(Object datapoint, int bufferSize) {
- checkForDeleted(bufferSize);
-
- long threshold;
- try {
- threshold = (Long) datapoint - granularity;
- } catch (ClassCastException e) {
- threshold = timestampWrapper.getTimestamp((DATA) datapoint) - granularity;
- }
-
- // return result
- return deleteAndCountExpired(threshold);
-
- }
-
- @Override
- public int notifyEviction(DATA datapoint, boolean triggered, int bufferSize) {
-
- checkForDeleted(bufferSize);
-
- // remember timestamp
- long time = timestampWrapper.getTimestamp(datapoint);
-
- // delete and count expired tuples
- long threshold = time - granularity;
- int counter = deleteAndCountExpired(threshold);
-
- // Add current element to buffer
- buffer.add(time);
-
- // return result
- return counter;
-
- }
-
- private void checkForDeleted(int bufferSize) {
- // check for deleted tuples (deletes by other policies)
- while (bufferSize < this.buffer.size()) {
- this.buffer.removeFirst();
- }
- }
-
- private int deleteAndCountExpired(long threshold) {
- int counter = 0;
- while (!buffer.isEmpty()) {
-
- if (buffer.getFirst() <= threshold) {
- buffer.removeFirst();
- counter++;
- } else {
- break;
- }
- }
- return counter;
-
- }
-
- @Override
- public TimeEvictionPolicy<DATA> clone() {
- return new TimeEvictionPolicy<DATA>(granularity, timestampWrapper);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicy.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicy.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicy.java
deleted file mode 100644
index 1e91b8e..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicy.java
+++ /dev/null
@@ -1,196 +0,0 @@
-/*
- * 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.policy;
-
-import java.util.LinkedList;
-
-import org.apache.flink.streaming.api.windowing.helper.Timestamp;
-import org.apache.flink.streaming.api.windowing.helper.TimestampWrapper;
-
-/**
- * This trigger policy triggers with regard to the time. The is measured using a
- * given {@link Timestamp} implementation. A point in time is always represented
- * as long. Therefore, parameters such as granularity and delay can be set as
- * long value as well.
- *
- * @param <DATA>
- * The type of the incoming data points which are processed by this
- * policy.
- */
-public class TimeTriggerPolicy<DATA> implements ActiveTriggerPolicy<DATA>,
- CloneableTriggerPolicy<DATA> {
-
- /**
- * auto generated version id
- */
- private static final long serialVersionUID = -5122753802440196719L;
-
- protected long startTime;
- protected long granularity;
- protected TimestampWrapper<DATA> timestampWrapper;
- protected long delay;
-
- /**
- * This trigger policy triggers with regard to the time. The is measured
- * using a given {@link Timestamp} implementation. A point in time is always
- * represented as long. Therefore, parameters such as granularity can be set
- * as long value as well. If this value for the granularity is set to 2 for
- * example, the policy will trigger at every second point in time.
- *
- * @param granularity
- * The granularity of the trigger. If this value is set to x the
- * policy will trigger at every x-th time point
- * @param timestampWrapper
- * The {@link TimestampWrapper} to measure the time with. This
- * can be either user defined of provided by the API.
- * @param timeWrapper
- * This policy creates fake elements to not miss windows in case
- * no element arrived within the duration of the window. This
- * extractor should wrap a long into such an element of type
- * DATA.
- */
- public TimeTriggerPolicy(long granularity, TimestampWrapper<DATA> timestampWrapper) {
- this(granularity, timestampWrapper, 0);
- }
-
- /**
- * This is mostly the same as
- * {@link TimeTriggerPolicy#TimeTriggerPolicy(long, Timestamp)}. In addition
- * to granularity and timestamp a delay can be specified for the first
- * trigger. If the start time given by the timestamp is x, the delay is y,
- * and the granularity is z, the first trigger will happen at x+y+z.
- *
- * @param granularity
- * The granularity of the trigger. If this value is set to 2 the
- * policy will trigger at every second time point
- * @param timestampWrapper
- * The {@link TimestampWrapper} to measure the time with. This
- * can be either user defined of provided by the API.
- * @param delay
- * A delay for the first trigger. If the start time given by the
- * timestamp is x, the delay is y, and the granularity is z, the
- * first trigger will happen at x+y+z.
- * @param timeWrapper
- * This policy creates fake elements to not miss windows in case
- * no element arrived within the duration of the window. This
- * extractor should wrap a long into such an element of type
- * DATA.
- */
- public TimeTriggerPolicy(long granularity, TimestampWrapper<DATA> timestampWrapper, long delay) {
- this.startTime = timestampWrapper.getStartTime() + delay;
- this.timestampWrapper = timestampWrapper;
- this.granularity = granularity;
- this.delay = delay;
- }
-
- /**
- * This method checks if we missed a window end. If this is the case we
- * trigger the missed windows using fake elements.
- */
- @Override
- public synchronized Object[] preNotifyTrigger(DATA datapoint) {
- LinkedList<Object> fakeElements = new LinkedList<Object>();
- // check if there is more then one window border missed
- // use > here. In case >= would fit, the regular call will do the job.
- while (timestampWrapper.getTimestamp(datapoint) >= startTime + granularity) {
- startTime += granularity;
- fakeElements.add(startTime - 1);
- }
- return (Object[]) fakeElements.toArray();
- }
-
- /**
- * In case {@link DefaultTimeStamp} is used, a runnable is returned which
- * triggers based on the current system time. If any other time measure is
- * used the method return null.
- *
- * @param callback
- * The object which is takes the callbacks for adding fake
- * elements out of the runnable.
- * @return A runnable is returned which triggers based on the current system
- * time. If any other time measure is used the method return null.
- */
- @Override
- public Runnable createActiveTriggerRunnable(ActiveTriggerCallback callback) {
- if (this.timestampWrapper.isDefaultTimestamp()) {
- return new TimeCheck(callback);
- } else {
- return null;
- }
- }
-
- /**
- * This method is only called in case the runnable triggers a window end
- * according to the {@link DefaultTimeStamp}.
- *
- * @param callback
- * The callback object.
- */
- private synchronized void activeFakeElementEmission(ActiveTriggerCallback callback) {
-
- // start time is excluded, but end time is included: >=
- if (System.currentTimeMillis() >= startTime + granularity) {
- startTime += granularity;
- callback.sendFakeElement(startTime - 1);
- }
-
- }
-
- private class TimeCheck implements Runnable {
- ActiveTriggerCallback callback;
-
- public TimeCheck(ActiveTriggerCallback callback) {
- this.callback = callback;
- }
-
- @Override
- public void run() {
- while (true) {
- // wait for the specified granularity
- try {
- Thread.sleep(granularity);
- } catch (InterruptedException e) {
- // ignore it...
- }
- // Trigger using the respective methods. Methods are
- // synchronized to prevent race conditions between real and fake
- // elements at the policy.
- activeFakeElementEmission(callback);
- }
- }
- }
-
- @Override
- public synchronized boolean notifyTrigger(DATA datapoint) {
- long recordTime = timestampWrapper.getTimestamp(datapoint);
- if (recordTime >= startTime + granularity) {
- if (granularity != 0) {
- startTime = recordTime - ((recordTime - startTime) % granularity);
- }
- return true;
- } else {
- return false;
- }
- }
-
- @Override
- public TimeTriggerPolicy<DATA> clone() {
- return new TimeTriggerPolicy<DATA>(granularity, timestampWrapper, delay);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TriggerPolicy.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TriggerPolicy.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TriggerPolicy.java
deleted file mode 100644
index c212df6..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TriggerPolicy.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * 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.policy;
-
-import java.io.Serializable;
-
-/**
- * Proves and returns if a new window should be started. In case the trigger
- * occurs (return value true) the UDF will be executed on the current element
- * buffer without the last added element which is provided as parameter. This
- * element will be added to the buffer after the execution of the UDF.
- *
- * @param <DATA>
- * The data type which can be handled by this policy
- */
-public interface TriggerPolicy<DATA> extends Serializable {
-
- /**
- * Proves and returns if a new window should be started. In case the trigger
- * occurs (return value true) the UDF will be executed on the current
- * element buffer without the last added element which is provided as
- * parameter. This element will be added to the buffer after the execution
- * of the UDF.
- *
- * There are possibly different strategies for eviction and triggering: 1)
- * including last data point: Better/faster for count eviction 2) excluding
- * last data point: Essentially required for time based eviction and delta
- * rules As 2) is required for some policies and the benefit of using 1) is
- * small for the others, policies are implemented according to 2).
- *
- * @param datapoint
- * the data point which arrived
- * @return true if the current windows should be closed, otherwise false. In
- * true case the given data point will be part of the next window
- * and will not be included in the current one.
- */
- public boolean notifyTrigger(DATA datapoint);
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TumblingEvictionPolicy.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TumblingEvictionPolicy.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TumblingEvictionPolicy.java
deleted file mode 100644
index 9a2e3c9..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TumblingEvictionPolicy.java
+++ /dev/null
@@ -1,90 +0,0 @@
-/*
- * 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.policy;
-
-/**
- * This eviction policy deletes all elements from the buffer in case a trigger
- * occurred. Therefore, it is the default eviction policy to be used for any
- * tumbling window.
- *
- * By default this policy does not react on fake elements. Wrap it in an
- * {@link ActiveEvictionPolicyWrapper} to make it clearing the buffer even on
- * fake elements.
- *
- * @param <DATA>
- * The type of the data points which is handled by this policy
- */
-public class TumblingEvictionPolicy<DATA> implements CloneableEvictionPolicy<DATA> {
-
- /**
- * Auto generated version ID
- */
- private static final long serialVersionUID = -4018019069267281155L;
-
- /**
- * Counter for the current number of elements in the buffer
- */
- private int counter = 0;
-
- /**
- * This is the default constructor providing no special functionality. This
- * eviction policy deletes all elements from the buffer in case a trigger
- * occurred. Therefore, it is the default eviction policy to be used for any
- * tumbling window.
- */
- public TumblingEvictionPolicy() {
- // default constructor, no further logic needed
- }
-
- /**
- * This constructor allows to set a custom start value for the element
- * counter.
- *
- * This eviction policy deletes all elements from the buffer in case a
- * trigger occurred. Therefore, it is the default eviction policy to be used
- * for any tumbling window.
- *
- * @param startValue
- * A start value for the element counter
- */
- public TumblingEvictionPolicy(int startValue) {
- this.counter = startValue;
- }
-
- /**
- * Deletes all elements from the buffer in case the trigger occurred.
- */
- @Override
- public int notifyEviction(Object datapoint, boolean triggered, int bufferSize) {
- if (triggered) {
- // The current data point will be part of the next window!
- // Therefore the counter needs to be set to one already.
- int tmpCounter = counter;
- counter = 1;
- return tmpCounter;
- } else {
- counter++;
- return 0;
- }
- }
-
- @Override
- public TumblingEvictionPolicy<DATA> clone() {
- return new TumblingEvictionPolicy<DATA>(counter);
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/io/BlockingQueueBroker.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/io/BlockingQueueBroker.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/io/BlockingQueueBroker.java
deleted file mode 100755
index 3ee2508..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/io/BlockingQueueBroker.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * 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.io;
-
-import java.util.concurrent.BlockingQueue;
-
-import org.apache.flink.runtime.iterative.concurrent.Broker;
-import org.apache.flink.streaming.api.streamrecord.StreamRecord;
-
-@SuppressWarnings("rawtypes")
-public class BlockingQueueBroker extends Broker<BlockingQueue<StreamRecord>> {
- /**
- * Singleton instance
- */
- private static final BlockingQueueBroker INSTANCE = new BlockingQueueBroker();
-
- private BlockingQueueBroker() {
- }
-
- /**
- * retrieve singleton instance
- */
- public static Broker<BlockingQueue<StreamRecord>> instance() {
- return INSTANCE;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/io/CoReaderIterator.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/io/CoReaderIterator.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/io/CoReaderIterator.java
deleted file mode 100755
index ed90c03..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/io/CoReaderIterator.java
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
- * 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.io;
-
-import java.io.IOException;
-
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.runtime.plugable.DeserializationDelegate;
-import org.apache.flink.runtime.plugable.ReusingDeserializationDelegate;
-
-/**
- * A CoReaderIterator wraps a {@link CoRecordReader} producing records of two
- * input types.
- */
-public class CoReaderIterator<T1, T2> {
-
- private final CoRecordReader<DeserializationDelegate<T1>, DeserializationDelegate<T2>> reader; // the
- // source
-
- protected final ReusingDeserializationDelegate<T1> delegate1;
- protected final ReusingDeserializationDelegate<T2> delegate2;
-
- public CoReaderIterator(
- CoRecordReader<DeserializationDelegate<T1>, DeserializationDelegate<T2>> reader,
- TypeSerializer<T1> serializer1, TypeSerializer<T2> serializer2) {
- this.reader = reader;
- this.delegate1 = new ReusingDeserializationDelegate<T1>(serializer1);
- this.delegate2 = new ReusingDeserializationDelegate<T2>(serializer2);
- }
-
- public int next(T1 target1, T2 target2) throws IOException {
- this.delegate1.setInstance(target1);
- this.delegate2.setInstance(target2);
-
- try {
- return this.reader.getNextRecord(this.delegate1, this.delegate2);
-
- } catch (InterruptedException e) {
- throw new IOException("Reader interrupted.", e);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/io/CoRecordReader.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/io/CoRecordReader.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/io/CoRecordReader.java
deleted file mode 100755
index 0b1b373..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/io/CoRecordReader.java
+++ /dev/null
@@ -1,221 +0,0 @@
-/*
- * 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.io;
-
-import org.apache.flink.core.io.IOReadableWritable;
-import org.apache.flink.runtime.event.task.TaskEvent;
-import org.apache.flink.runtime.io.network.api.reader.BufferReaderBase;
-import org.apache.flink.runtime.io.network.api.reader.MutableRecordReader;
-import org.apache.flink.runtime.io.network.api.reader.ReaderBase;
-import org.apache.flink.runtime.io.network.api.serialization.AdaptiveSpanningRecordDeserializer;
-import org.apache.flink.runtime.io.network.api.serialization.RecordDeserializer;
-import org.apache.flink.runtime.io.network.buffer.Buffer;
-import org.apache.flink.runtime.util.event.EventListener;
-
-import java.io.IOException;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
-
-/**
- * A CoRecordReader wraps {@link MutableRecordReader}s of two different input
- * types to read records effectively.
- */
-@SuppressWarnings("rawtypes")
-public class CoRecordReader<T1 extends IOReadableWritable, T2 extends IOReadableWritable> implements ReaderBase, EventListener<BufferReaderBase> {
-
- private final BufferReaderBase bufferReader1;
-
- private final BufferReaderBase bufferReader2;
-
- private final BlockingQueue<Integer> availableRecordReaders = new LinkedBlockingQueue<Integer>();
-
- private AdaptiveSpanningRecordDeserializer[] reader1RecordDeserializers;
-
- private RecordDeserializer<T1> reader1currentRecordDeserializer;
-
- private AdaptiveSpanningRecordDeserializer[] reader2RecordDeserializers;
-
- private RecordDeserializer<T2> reader2currentRecordDeserializer;
-
- // 0 => none, 1 => reader (T1), 2 => reader (T2)
- private int currentReaderIndex;
-
- private boolean hasRequestedPartitions;
-
- public CoRecordReader(BufferReaderBase bufferReader1, BufferReaderBase bufferReader2) {
- this.bufferReader1 = bufferReader1;
- this.bufferReader2 = bufferReader2;
-
- this.reader1RecordDeserializers = new AdaptiveSpanningRecordDeserializer[bufferReader1.getNumberOfInputChannels()];
- this.reader2RecordDeserializers = new AdaptiveSpanningRecordDeserializer[bufferReader2.getNumberOfInputChannels()];
-
- for (int i = 0; i < reader1RecordDeserializers.length; i++) {
- reader1RecordDeserializers[i] = new AdaptiveSpanningRecordDeserializer<T1>();
- }
-
- for (int i = 0; i < reader2RecordDeserializers.length; i++) {
- reader2RecordDeserializers[i] = new AdaptiveSpanningRecordDeserializer<T2>();
- }
-
- bufferReader1.subscribeToReader(this);
- bufferReader2.subscribeToReader(this);
- }
-
- public void requestPartitionsOnce() throws IOException {
- if (!hasRequestedPartitions) {
- bufferReader1.requestPartitionsOnce();
- bufferReader2.requestPartitionsOnce();
-
- hasRequestedPartitions = true;
- }
- }
-
- @SuppressWarnings("unchecked")
- protected int getNextRecord(T1 target1, T2 target2) throws IOException, InterruptedException {
-
- requestPartitionsOnce();
-
- while (true) {
- if (currentReaderIndex == 0) {
- if ((bufferReader1.isFinished() && bufferReader2.isFinished())) {
- return 0;
- }
-
- currentReaderIndex = getNextReaderIndexBlocking();
- }
-
- if (currentReaderIndex == 1) {
- while (true) {
- if (reader1currentRecordDeserializer != null) {
- RecordDeserializer.DeserializationResult result = reader1currentRecordDeserializer.getNextRecord(target1);
-
- if (result.isBufferConsumed()) {
- reader1currentRecordDeserializer.getCurrentBuffer().recycle();
- reader1currentRecordDeserializer = null;
-
- currentReaderIndex = 0;
- }
-
- if (result.isFullRecord()) {
- return 1;
- }
- }
-
- final Buffer nextBuffer = bufferReader1.getNextBufferBlocking();
- final int channelIndex = bufferReader1.getChannelIndexOfLastBuffer();
-
- if (nextBuffer == null) {
- currentReaderIndex = 0;
-
- break;
- }
-
- reader1currentRecordDeserializer = reader1RecordDeserializers[channelIndex];
- reader1currentRecordDeserializer.setNextBuffer(nextBuffer);
- }
- }
- else if (currentReaderIndex == 2) {
- while (true) {
- if (reader2currentRecordDeserializer != null) {
- RecordDeserializer.DeserializationResult result = reader2currentRecordDeserializer.getNextRecord(target2);
-
- if (result.isBufferConsumed()) {
- reader2currentRecordDeserializer.getCurrentBuffer().recycle();
- reader2currentRecordDeserializer = null;
-
- currentReaderIndex = 0;
- }
-
- if (result.isFullRecord()) {
- return 2;
- }
- }
-
- final Buffer nextBuffer = bufferReader2.getNextBufferBlocking();
- final int channelIndex = bufferReader2.getChannelIndexOfLastBuffer();
-
- if (nextBuffer == null) {
- currentReaderIndex = 0;
-
- break;
- }
-
- reader2currentRecordDeserializer = reader2RecordDeserializers[channelIndex];
- reader2currentRecordDeserializer.setNextBuffer(nextBuffer);
- }
- }
- else {
- throw new IllegalStateException("Bug: unexpected current reader index.");
- }
- }
- }
-
- private int getNextReaderIndexBlocking() throws InterruptedException {
- return availableRecordReaders.take();
- }
-
- // ------------------------------------------------------------------------
- // Data availability notifications
- // ------------------------------------------------------------------------
-
- @Override
- public void onEvent(BufferReaderBase bufferReader) {
- if (bufferReader == bufferReader1) {
- availableRecordReaders.add(1);
- }
- else if (bufferReader == bufferReader2) {
- availableRecordReaders.add(2);
- }
- }
-
- // ------------------------------------------------------------------------
-
- @Override
- public boolean isFinished() {
- return bufferReader1.isFinished() && bufferReader2.isFinished();
- }
-
- @Override
- public void subscribeToTaskEvent(EventListener<TaskEvent> eventListener, Class<? extends TaskEvent> eventType) {
- bufferReader1.subscribeToTaskEvent(eventListener, eventType);
- bufferReader2.subscribeToTaskEvent(eventListener, eventType);
- }
-
- @Override
- public void sendTaskEvent(TaskEvent event) throws IOException, InterruptedException {
- bufferReader1.sendTaskEvent(event);
- bufferReader2.sendTaskEvent(event);
- }
-
- @Override
- public void setIterativeReader() {
- bufferReader1.setIterativeReader();
- bufferReader2.setIterativeReader();
- }
-
- @Override
- public void startNextSuperstep() {
- bufferReader1.startNextSuperstep();
- bufferReader2.startNextSuperstep();
- }
-
- @Override
- public boolean hasReachedEndOfSuperstep() {
- return bufferReader1.hasReachedEndOfSuperstep() && bufferReader2.hasReachedEndOfSuperstep();
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/io/StreamRecordWriter.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/io/StreamRecordWriter.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/io/StreamRecordWriter.java
deleted file mode 100755
index b7af589..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/io/StreamRecordWriter.java
+++ /dev/null
@@ -1,83 +0,0 @@
-/*
- * 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.io;
-
-import org.apache.flink.core.io.IOReadableWritable;
-import org.apache.flink.runtime.io.network.api.writer.BufferWriter;
-import org.apache.flink.runtime.io.network.api.writer.ChannelSelector;
-import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
-import org.apache.flink.runtime.io.network.api.writer.RoundRobinChannelSelector;
-
-import java.io.IOException;
-
-public class StreamRecordWriter<T extends IOReadableWritable> extends RecordWriter<T> {
-
- private long timeout;
-
- private OutputFlusher outputFlusher;
-
- public StreamRecordWriter(BufferWriter writer) {
- this(writer, new RoundRobinChannelSelector<T>(), 1000);
- }
-
- public StreamRecordWriter(BufferWriter writer, ChannelSelector<T> channelSelector) {
- this(writer, channelSelector, 1000);
- }
-
- public StreamRecordWriter(BufferWriter writer, ChannelSelector<T> channelSelector, long timeout) {
- super(writer, channelSelector);
-
- this.timeout = timeout;
- this.outputFlusher = new OutputFlusher();
-
- outputFlusher.start();
- }
-
- public void close() {
- try {
- if (outputFlusher != null) {
- outputFlusher.terminate();
- outputFlusher.join();
- }
-
- flush();
- } catch (IOException e) {
- e.printStackTrace();
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- }
-
- private class OutputFlusher extends Thread {
- private boolean running = true;
- public void terminate() {
- running = false;
- }
- @Override
- public void run() {
- while (running && !writer.isFinished()) {
- try {
- flush();
- Thread.sleep(timeout);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/BroadcastPartitioner.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/BroadcastPartitioner.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/BroadcastPartitioner.java
deleted file mode 100644
index 3813c8a..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/BroadcastPartitioner.java
+++ /dev/null
@@ -1,55 +0,0 @@
-/*
- * 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.partitioner;
-
-import org.apache.flink.runtime.plugable.SerializationDelegate;
-import org.apache.flink.streaming.api.streamrecord.StreamRecord;
-
-/**
- * Partitioner that selects all the output channels.
- *
- * @param <T>
- * Type of the Tuple
- */
-public class BroadcastPartitioner<T> extends StreamPartitioner<T> {
- private static final long serialVersionUID = 1L;
-
- int[] returnArray;
- boolean set;
- int setNumber;
-
- public BroadcastPartitioner() {
- super(PartitioningStrategy.BROADCAST);
- }
-
- @Override
- public int[] selectChannels(SerializationDelegate<StreamRecord<T>> record,
- int numberOfOutputChannels) {
- if (set && setNumber == numberOfOutputChannels) {
- return returnArray;
- } else {
- this.returnArray = new int[numberOfOutputChannels];
- for (int i = 0; i < numberOfOutputChannels; i++) {
- returnArray[i] = i;
- }
- set = true;
- setNumber = numberOfOutputChannels;
- return returnArray;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/DistributePartitioner.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/DistributePartitioner.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/DistributePartitioner.java
deleted file mode 100644
index 7299397..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/DistributePartitioner.java
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- * 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.partitioner;
-
-import org.apache.flink.runtime.plugable.SerializationDelegate;
-import org.apache.flink.streaming.api.streamrecord.StreamRecord;
-
-/**
- * Partitioner that distributes the data equally by cycling through the output
- * channels.
- *
- * @param <T>
- * Type of the Tuple
- */
-public class DistributePartitioner<T> extends StreamPartitioner<T> {
- private static final long serialVersionUID = 1L;
-
- private int[] returnArray = new int[] {-1};
-
- public DistributePartitioner(boolean forward) {
- super(forward ? PartitioningStrategy.FORWARD : PartitioningStrategy.DISTRIBUTE);
- }
-
- @Override
- public int[] selectChannels(SerializationDelegate<StreamRecord<T>> record,
- int numberOfOutputChannels) {
- this.returnArray[0] = (this.returnArray[0] + 1) % numberOfOutputChannels;
-
- return this.returnArray;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/FieldsPartitioner.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/FieldsPartitioner.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/FieldsPartitioner.java
deleted file mode 100644
index 6bc036e..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/FieldsPartitioner.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * 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.partitioner;
-
-import org.apache.flink.api.java.functions.KeySelector;
-import org.apache.flink.runtime.plugable.SerializationDelegate;
-import org.apache.flink.streaming.api.streamrecord.StreamRecord;
-
-/**
- * Partitioner that selects the same (one) channel for two Tuples having a
- * specified fields equal.
- *
- * @param <T>
- * Type of the Tuple
- */
-public class FieldsPartitioner<T> extends StreamPartitioner<T> {
- private static final long serialVersionUID = 1L;
-
- private int[] returnArray = new int[1];;
- KeySelector<T, ?> keySelector;
-
- public FieldsPartitioner(KeySelector<T, ?> keySelector) {
- super(PartitioningStrategy.GROUPBY);
- this.keySelector = keySelector;
- }
-
- @Override
- public int[] selectChannels(SerializationDelegate<StreamRecord<T>> record,
- int numberOfOutputChannels) {
- returnArray[0] = Math.abs(record.getInstance().getKey(keySelector).hashCode()
- % numberOfOutputChannels);
-
- return returnArray;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/GlobalPartitioner.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/GlobalPartitioner.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/GlobalPartitioner.java
deleted file mode 100644
index c73ca71..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/GlobalPartitioner.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * 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.partitioner;
-
-import org.apache.flink.runtime.plugable.SerializationDelegate;
-import org.apache.flink.streaming.api.streamrecord.StreamRecord;
-
-//Group to the partitioner with the lowest id
-public class GlobalPartitioner<T> extends StreamPartitioner<T> {
- private static final long serialVersionUID = 1L;
-
- private int[] returnArray = new int[] { 0 };
-
- public GlobalPartitioner() {
- super(PartitioningStrategy.GLOBAL);
- }
-
- @Override
- public int[] selectChannels(SerializationDelegate<StreamRecord<T>> record,
- int numberOfOutputChannels) {
- return returnArray;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/ShufflePartitioner.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/ShufflePartitioner.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/ShufflePartitioner.java
deleted file mode 100644
index 318de3f..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/ShufflePartitioner.java
+++ /dev/null
@@ -1,49 +0,0 @@
-/*
- * 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.partitioner;
-
-import java.util.Random;
-
-import org.apache.flink.runtime.plugable.SerializationDelegate;
-import org.apache.flink.streaming.api.streamrecord.StreamRecord;
-
-/**
- * Partitioner that distributes the data equally by selecting one output channel
- * randomly.
- *
- * @param <T>
- * Type of the Tuple
- */
-public class ShufflePartitioner<T> extends StreamPartitioner<T> {
- private static final long serialVersionUID = 1L;
-
- private Random random = new Random();
-
- private int[] returnArray = new int[1];
-
- public ShufflePartitioner() {
- super(PartitioningStrategy.SHUFFLE);
- }
-
- @Override
- public int[] selectChannels(SerializationDelegate<StreamRecord<T>> record,
- int numberOfOutputChannels) {
- returnArray[0] = random.nextInt(numberOfOutputChannels);
- return returnArray;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/StreamPartitioner.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/StreamPartitioner.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/StreamPartitioner.java
deleted file mode 100644
index 19a8dba..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/StreamPartitioner.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * 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.partitioner;
-
-import java.io.Serializable;
-
-import org.apache.flink.runtime.io.network.api.writer.ChannelSelector;
-import org.apache.flink.runtime.plugable.SerializationDelegate;
-import org.apache.flink.streaming.api.streamrecord.StreamRecord;
-
-public abstract class StreamPartitioner<T> implements
- ChannelSelector<SerializationDelegate<StreamRecord<T>>>, Serializable {
-
- public enum PartitioningStrategy {
-
- FORWARD, DISTRIBUTE, SHUFFLE, BROADCAST, GLOBAL, GROUPBY;
-
- }
-
- private static final long serialVersionUID = 1L;
- private PartitioningStrategy strategy;
-
- public StreamPartitioner(PartitioningStrategy strategy) {
- this.strategy = strategy;
- }
-
- public PartitioningStrategy getStrategy() {
- return strategy;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/CircularFifoList.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/CircularFifoList.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/CircularFifoList.java
deleted file mode 100644
index 02a628b..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/CircularFifoList.java
+++ /dev/null
@@ -1,112 +0,0 @@
-/*
- * 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.state;
-
-import java.io.Serializable;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Queue;
-
-/**
- * A simple class, that manages a circular queue with sliding interval. If the
- * queue if full and a new element is added, the elements that belong to the
- * first sliding interval are removed.
- */
-public class CircularFifoList<T> implements Serializable {
- private static final long serialVersionUID = 1L;
-
- private Queue<T> queue;
- private Queue<Long> slideSizes;
- private long counter;
- private Iterable<T> iterable;
-
- public CircularFifoList() {
- this.queue = new LinkedList<T>();
- this.slideSizes = new LinkedList<Long>();
- this.counter = 0;
- this.iterable = new ListIterable();
- }
-
- public void add(T element) {
- queue.add(element);
- counter++;
- }
-
- public void newSlide() {
- slideSizes.add(counter);
- counter = 0;
- }
-
- public void shiftWindow() {
- shiftWindow(1);
- }
-
- public void shiftWindow(int numberOfSlides) {
-
- if (numberOfSlides <= slideSizes.size()) {
- for (int i = 0; i < numberOfSlides; i++) {
- Long firstSlideSize = slideSizes.remove();
-
- for (int j = 0; j < firstSlideSize; j++) {
- queue.remove();
- }
- }
- } else {
- slideSizes.clear();
- queue.clear();
- counter = 0;
- }
-
- }
-
- @SuppressWarnings("unchecked")
- public List<T> getElements(){
- return (List<T>) queue;
- }
-
- public Iterator<T> getIterator() {
- return queue.iterator();
- }
-
- public Iterable<T> getIterable() {
- return iterable;
- }
-
- private class ListIterable implements Iterable<T>, Serializable {
-
- private static final long serialVersionUID = 1L;
-
- @Override
- public Iterator<T> iterator() {
- return getIterator();
- }
-
- }
-
- public boolean isEmpty() {
- return queue.isEmpty();
- }
-
- @Override
- public String toString() {
- return queue.toString();
- }
-
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/GraphState.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/GraphState.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/GraphState.java
deleted file mode 100644
index 945c0fd..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/GraphState.java
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
- * 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.state;
-
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Set;
-
-/**
- * GraphState represents a special {@link MapState} for storing graph
- * structures.
- *
- */
-public class GraphState extends MapState<Integer, Set<Integer>> {
-
- private static final long serialVersionUID = 1L;
-
- public GraphState() {
- state = new HashMap<Integer, Set<Integer>>();
- }
-
- public void insertDirectedEdge(int sourceNode, int targetNode) {
- if (!containsKey(sourceNode)) {
- state.put(sourceNode, new HashSet<Integer>());
- }
- state.get(sourceNode).add(targetNode);
- updatedItems.add(sourceNode);
- removedItems.remove(sourceNode);
- }
-
- public void insertUndirectedEdge(int sourceNode, int targetNode) {
- if (!state.containsKey(sourceNode)) {
- state.put(sourceNode, new HashSet<Integer>());
- }
- if (!state.containsKey(targetNode)) {
- state.put(targetNode, new HashSet<Integer>());
- }
- state.get(sourceNode).add(targetNode);
- state.get(targetNode).add(sourceNode);
-
- updatedItems.add(sourceNode);
- removedItems.remove(sourceNode);
- updatedItems.add(targetNode);
- removedItems.remove(targetNode);
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/MapState.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/MapState.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/MapState.java
deleted file mode 100644
index 85aec52..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/MapState.java
+++ /dev/null
@@ -1,201 +0,0 @@
-/*
- * 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.state;
-
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-
-import org.apache.flink.streaming.state.checkpoint.MapCheckpoint;
-import org.apache.flink.streaming.state.checkpoint.StateCheckpoint;
-
-/**
- * A Map that can be used as a partitionable operator state, for both fault
- * tolerance and rebalancing in Flink programs. The default repartitioner for
- * this operator repartitions by the hashcodes of the keys in the map. </br>
- * </br> The MapState also allows for incremental (data efficient) checkpointing
- * of the state. The entries in the map should only be modified by using the
- * dedicated methods: {@link #put(Object, Object)},{@link #remove(Object)},
- * {@link #putAll(Map)} and {@link #clear}. Directly modifying the entryset
- * will cause errors when checkpointing.
- *
- * @param <K>
- * The type of the keys.
- * @param <V>
- * The type of the values.
- */
-public class MapState<K, V> extends PartitionableState<Map<K, V>> implements Map<K, V> {
-
- private static final long serialVersionUID = 1L;
-
- protected Set<K> removedItems;
- protected Set<K> updatedItems;
- protected boolean clear;
-
- public MapState() {
- this.state = new HashMap<K, V>();
- this.updatedItems = new HashSet<K>();
- this.removedItems = new HashSet<K>();
- }
-
- @Override
- public StateCheckpoint<Map<K, V>> checkpoint() {
- this.updatedItems.removeAll(removedItems);
- StateCheckpoint<Map<K, V>> checkpoint = new MapCheckpoint<K, V>(this);
- resetHistory();
- return checkpoint;
- }
-
- @Override
- public OperatorState<Map<K, V>> restore(StateCheckpoint<Map<K, V>> checkpoint) {
- this.state = checkpoint.getCheckpointedState();
- resetHistory();
- return this;
- }
-
- @Override
- public OperatorState<Map<K, V>>[] repartition(int numberOfPartitions) {
- @SuppressWarnings("unchecked")
- MapState<K, V>[] states = new MapState[numberOfPartitions];
- for (int i = 0; i < numberOfPartitions; i++) {
- states[i] = new MapState<K, V>();
- }
- for (Entry<K, V> entry : this.state.entrySet()) {
- int partition = Math.abs(entry.getKey().hashCode() % numberOfPartitions);
- states[partition].state.put(entry.getKey(), entry.getValue());
- }
- return states;
- }
-
- @Override
- public OperatorState<Map<K, V>> reBuild(OperatorState<Map<K, V>>... parts) {
- clear();
- for (OperatorState<Map<K, V>> operatorState : parts) {
- putAll(operatorState.state);
- }
- return this;
- }
-
- public void resetHistory() {
- clear = false;
- removedItems.clear();
- updatedItems.clear();
- }
-
- @Override
- public int size() {
- return state.size();
- }
-
- @Override
- public boolean isEmpty() {
- return state.isEmpty();
- }
-
- @Override
- public boolean containsKey(Object key) {
- return state.containsKey(key);
- }
-
- @Override
- public boolean containsValue(Object value) {
- return state.containsValue(value);
- }
-
- @Override
- public V get(Object key) {
- return state.get(key);
- }
-
- @Override
- public V put(K key, V value) {
- updatedItems.add(key);
- removedItems.remove(key);
- return state.put(key, value);
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public V remove(Object key) {
- V removed = state.remove(key);
- if (removed != null) {
- removedItems.add((K) key);
- updatedItems.remove((K) key);
- }
-
- return removed;
- }
-
- @Override
- public void putAll(Map<? extends K, ? extends V> m) {
- for (K key : m.keySet()) {
- updatedItems.add(key);
- removedItems.remove(key);
- }
- state.putAll(m);
- }
-
- @Override
- public void clear() {
- clear = true;
- updatedItems.clear();
- removedItems.clear();
- state.clear();
- }
-
- @Override
- public Set<K> keySet() {
- return state.keySet();
- }
-
- @Override
- public Collection<V> values() {
- return state.values();
- }
-
- @Override
- public Set<Entry<K, V>> entrySet() {
- return state.entrySet();
- }
-
- @Override
- public Map<K, V> getState() {
- throw new RuntimeException(
- "State object should be accessed using the Map interface provided by the MapState");
- }
-
- @Override
- public OperatorState<Map<K, V>> setState(Map<K, V> state) {
- throw new RuntimeException(
- "State object should be accessed using the Map interface provided by the MapState");
- }
-
- public Set<K> getRemovedItems() {
- return removedItems;
- }
-
- public Set<K> getUpdatedItems() {
- return updatedItems;
- }
-
- public boolean isCleared() {
- return clear;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/NullableCircularBuffer.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/NullableCircularBuffer.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/NullableCircularBuffer.java
deleted file mode 100644
index aa2a1bc..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/NullableCircularBuffer.java
+++ /dev/null
@@ -1,362 +0,0 @@
-/*
- * 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.state;
-
-import java.io.IOException;
-import java.io.ObjectInputStream;
-import java.io.ObjectOutputStream;
-import java.io.Serializable;
-import java.util.AbstractCollection;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Iterator;
-import java.util.NoSuchElementException;
-
-import org.apache.commons.collections.BoundedCollection;
-import org.apache.commons.collections.Buffer;
-import org.apache.commons.collections.BufferUnderflowException;
-
-@SuppressWarnings("rawtypes")
-public class NullableCircularBuffer extends AbstractCollection implements Buffer,
- BoundedCollection, Serializable {
-
- /** Serialization version */
- private static final long serialVersionUID = 5603722811189451017L;
-
- /** Underlying storage array */
- private transient Object[] elements;
-
- /** Array index of first (oldest) buffer element */
- private transient int start = 0;
-
- /**
- * Index mod maxElements of the array position following the last buffer
- * element. Buffer elements start at elements[start] and "wrap around"
- * elements[maxElements-1], ending at elements[decrement(end)]. For example,
- * elements = {c,a,b}, start=1, end=1 corresponds to the buffer [a,b,c].
- */
- private transient int end = 0;
-
- /** Flag to indicate if the buffer is currently full. */
- private transient boolean full = false;
-
- /** Capacity of the buffer */
- private final int maxElements;
-
- /**
- * Constructs a new <code>BoundedFifoBuffer</code> big enough to hold 32
- * elements.
- */
- public NullableCircularBuffer() {
- this(32);
- }
-
- /**
- * Constructs a new <code>BoundedFifoBuffer</code> big enough to hold the
- * specified number of elements.
- *
- * @param size
- * the maximum number of elements for this fifo
- * @throws IllegalArgumentException
- * if the size is less than 1
- */
- public NullableCircularBuffer(int size) {
- if (size <= 0) {
- throw new IllegalArgumentException("The size must be greater than 0");
- }
- elements = new Object[size];
- maxElements = elements.length;
- }
-
- /**
- * Constructs a new <code>BoundedFifoBuffer</code> big enough to hold all of
- * the elements in the specified collection. That collection's elements will
- * also be added to the buffer.
- *
- * @param coll
- * the collection whose elements to add, may not be null
- * @throws NullPointerException
- * if the collection is null
- */
- @SuppressWarnings("unchecked")
- public NullableCircularBuffer(Collection coll) {
- this(coll.size());
- addAll(coll);
- }
-
- // -----------------------------------------------------------------------
- /**
- * Write the buffer out using a custom routine.
- *
- * @param out
- * the output stream
- * @throws IOException
- */
- private void writeObject(ObjectOutputStream out) throws IOException {
- out.defaultWriteObject();
- out.writeInt(size());
- for (Iterator it = iterator(); it.hasNext();) {
- out.writeObject(it.next());
- }
- }
-
- /**
- * Read the buffer in using a custom routine.
- *
- * @param in
- * the input stream
- * @throws IOException
- * @throws ClassNotFoundException
- */
- private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
- in.defaultReadObject();
- elements = new Object[maxElements];
- int size = in.readInt();
- for (int i = 0; i < size; i++) {
- elements[i] = in.readObject();
- }
- start = 0;
- full = (size == maxElements);
- if (full) {
- end = 0;
- } else {
- end = size;
- }
- }
-
- // -----------------------------------------------------------------------
- /**
- * Returns the number of elements stored in the buffer.
- *
- * @return this buffer's size
- */
- public int size() {
- int size = 0;
-
- if (end < start) {
- size = maxElements - start + end;
- } else if (end == start) {
- size = (full ? maxElements : 0);
- } else {
- size = end - start;
- }
-
- return size;
- }
-
- /**
- * Returns true if this buffer is empty; false otherwise.
- *
- * @return true if this buffer is empty
- */
- public boolean isEmpty() {
- return size() == 0;
- }
-
- /**
- * Returns true if this collection is full and no new elements can be added.
- *
- * @return <code>true</code> if the collection is full
- */
- public boolean isFull() {
- return size() == maxElements;
- }
-
- /**
- * Gets the maximum size of the collection (the bound).
- *
- * @return the maximum number of elements the collection can hold
- */
- public int maxSize() {
- return maxElements;
- }
-
- /**
- * Clears this buffer.
- */
- public void clear() {
- full = false;
- start = 0;
- end = 0;
- Arrays.fill(elements, null);
- }
-
- /**
- * Adds the given element to this buffer.
- *
- * @param element
- * the element to add
- * @return true, always
- */
- public boolean add(Object element) {
-
- if (isFull()) {
- remove();
- }
-
- elements[end++] = element;
-
- if (end >= maxElements) {
- end = 0;
- }
-
- if (end == start) {
- full = true;
- }
-
- return true;
- }
-
- /**
- * Returns the least recently inserted element in this buffer.
- *
- * @return the least recently inserted element
- * @throws BufferUnderflowException
- * if the buffer is empty
- */
- public Object get() {
- if (isEmpty()) {
- throw new BufferUnderflowException("The buffer is already empty");
- }
-
- return elements[start];
- }
-
- /**
- * Removes the least recently inserted element from this buffer.
- *
- * @return the least recently inserted element
- * @throws BufferUnderflowException
- * if the buffer is empty
- */
- public Object remove() {
- if (isEmpty()) {
- throw new BufferUnderflowException("The buffer is already empty");
- }
-
- Object element = elements[start];
-
- elements[start++] = null;
-
- if (start >= maxElements) {
- start = 0;
- }
-
- full = false;
-
- return element;
- }
-
- /**
- * Increments the internal index.
- *
- * @param index
- * the index to increment
- * @return the updated index
- */
- private int increment(int index) {
- index++;
- if (index >= maxElements) {
- index = 0;
- }
- return index;
- }
-
- /**
- * Decrements the internal index.
- *
- * @param index
- * the index to decrement
- * @return the updated index
- */
- private int decrement(int index) {
- index--;
- if (index < 0) {
- index = maxElements - 1;
- }
- return index;
- }
-
- /**
- * Returns an iterator over this buffer's elements.
- *
- * @return an iterator over this buffer's elements
- */
- public Iterator iterator() {
- return new Iterator() {
-
- private int index = start;
- private int lastReturnedIndex = -1;
- private boolean isFirst = full;
-
- public boolean hasNext() {
- return isFirst || (index != end);
-
- }
-
- public Object next() {
- if (!hasNext()) {
- throw new NoSuchElementException();
- }
- isFirst = false;
- lastReturnedIndex = index;
- index = increment(index);
- return elements[lastReturnedIndex];
- }
-
- public void remove() {
- if (lastReturnedIndex == -1) {
- throw new IllegalStateException();
- }
-
- // First element can be removed quickly
- if (lastReturnedIndex == start) {
- NullableCircularBuffer.this.remove();
- lastReturnedIndex = -1;
- return;
- }
-
- int pos = lastReturnedIndex + 1;
- if (start < lastReturnedIndex && pos < end) {
- // shift in one part
- System.arraycopy(elements, pos, elements, lastReturnedIndex, end - pos);
- } else {
- // Other elements require us to shift the subsequent
- // elements
- while (pos != end) {
- if (pos >= maxElements) {
- elements[pos - 1] = elements[0];
- pos = 0;
- } else {
- elements[decrement(pos)] = elements[pos];
- pos = increment(pos);
- }
- }
- }
-
- lastReturnedIndex = -1;
- end = decrement(end);
- elements[end] = null;
- full = false;
- index = decrement(index);
- }
-
- };
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/OperatorState.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/OperatorState.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/OperatorState.java
deleted file mode 100644
index a0cedba..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/OperatorState.java
+++ /dev/null
@@ -1,105 +0,0 @@
-/*
- * 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.state;
-
-import java.io.Serializable;
-
-import org.apache.flink.streaming.state.checkpoint.StateCheckpoint;
-
-/**
- * Abstract class for representing operator states in Flink programs. By
- * implementing the methods declared in this abstraction the state of the
- * operator can be checkpointed by the fault tolerance mechanism.
- *
- * @param <T>
- * The type of the operator state.
- */
-public abstract class OperatorState<T> implements Serializable {
-
- private static final long serialVersionUID = 1L;
-
- protected T state;
-
- /**
- * Constructor used for initializing the state. In case of failure, the
- * state will be reinitialized using this constructor, then
- * {@link #restore(StateCheckpoint)} will be used to restore from the last
- * available backup.
- */
- public OperatorState() {
- state = null;
- }
-
- /**
- * Initializes the state using the given state object.
- *
- * @param initialState
- * The initial state object
- */
- public OperatorState(T initialState) {
- state = initialState;
- }
-
- /**
- * Returns the currently stored state object.
- *
- * @return The state.
- */
- public T getState() {
- return state;
- }
-
- /**
- * Sets the current state object.
- *
- * @param state
- * The new state object.
- * @return The operator state with the new state object set.
- */
- public OperatorState<T> setState(T state) {
- this.state = state;
- return this;
- }
-
- /**
- * Creates a {@link StateCheckpoint} that will be used to backup the state
- * for failure recovery.
- *
- * @return The {@link StateCheckpoint} created.
- */
- public abstract StateCheckpoint<T> checkpoint();
-
- /**
- * Restores the state from the given {@link StateCheckpoint}.
- *
- * @param checkpoint
- * The checkpoint to restore from
- * @return The restored operator.
- */
- public abstract OperatorState<T> restore(StateCheckpoint<T> checkpoint);
-
- @Override
- public String toString() {
- return state.toString();
- }
-
- public boolean stateEquals(OperatorState<T> other) {
- return state.equals(other.state);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/PartitionableState.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/PartitionableState.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/PartitionableState.java
deleted file mode 100644
index c58c545..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/PartitionableState.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * 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.state;
-
-/**
- * Base class for representing operator states that can be repartitioned for
- * state state and load balancing.
- *
- * @param <T>
- * The type of the operator state.
- */
-public abstract class PartitionableState<T> extends OperatorState<T> {
-
- private static final long serialVersionUID = 1L;
-
- /**
- * Repartitions(divides) the current state into the given number of new
- * partitions. The created partitions will be used to redistribute then
- * rebuild the state among the parallel instances of the operator. The
- * implementation should reflect the partitioning of the input values to
- * maintain correct operator behavior.
- *
- * </br> </br> It is also assumed that if we would {@link #reBuild} the
- * repartitioned state we would basically get the same as before.
- *
- *
- * @param numberOfPartitions
- * The desired number of partitions. The method must return an
- * array of that size.
- * @return The array containing the state part for each partition.
- */
- public abstract OperatorState<T>[] repartition(int numberOfPartitions);
-
- /**
- * Rebuilds the current state partition from the given parts. Used for
- * building the state after a re-balance phase.
- *
- * @param parts
- * The state parts that will be used to rebuild the current
- * partition.
- * @return The rebuilt operator state.
- */
- public abstract OperatorState<T> reBuild(OperatorState<T>... parts);
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/SimpleState.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/SimpleState.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/SimpleState.java
deleted file mode 100644
index 7ae1f81..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/state/SimpleState.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * 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.state;
-
-import org.apache.flink.streaming.state.checkpoint.StateCheckpoint;
-
-/**
- * Basic {@link OperatorState} for storing and updating simple objects. By default the
- * whole stored object is checkpointed at each backup. Override checkpoint to
- * allow a more fine-grained behavior.
- *
- * @param <T>
- * The type of the stored state object.
- */
-public class SimpleState<T> extends OperatorState<T> {
-
- private static final long serialVersionUID = 1L;
-
- public SimpleState() {
- super();
- }
-
- public SimpleState(T initialState) {
- super(initialState);
- }
-
- @Override
- public StateCheckpoint<T> checkpoint() {
- return new StateCheckpoint<T>(this);
- }
-
- @Override
- public OperatorState<T> restore(StateCheckpoint<T> checkpoint) {
- this.state = checkpoint.getCheckpointedState();
- return this;
- }
-
-}