[streaming] New Source and state checkpointing interfaces that allow...

[streaming] New Source and state checkpointing interfaces that allow operations to interact with the state checkpointing in a more precise manner.

flink-core/src/main/java/org/apache/flink/api/common/functions/Function.java

@@ -19,9 +19,10 @@
 package org.apache.flink.api.common.functions;
 
 /**
- * An base interface for all user-defined functions. This interface is empty in order
- * to enable functions that are SAM (single abstract method) interfaces, so that they
- * can be called as Java 8 lambdas
+ * The base interface for all user-defined functions.
+ * 
+ * <p>This interface is empty in order to allow extending interfaces to
+ * be SAM (single abstract method) interfaces that can be implemented via Java 8 lambdas.</p>
  */
 public interface Function {
 }

flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/checkpoint/CheckpointCommitter.java

+/*
+ * 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.checkpoint;
+
+/**
+ * This interface must be implemented by functions/operations that want to receive
+ * a commit notification once a checkpoint has been completely acknowledged by all
+ * participants.
+ */
+public interface CheckpointCommitter {
+
+	/**
+	 * This method is called as a notification once a distributed checkpoint has been completed.
+	 * 
+	 * Note that any exception during this method will not cause the checkpoint to
+	 * fail any more.
+	 * 
+	 * @param checkpointId The ID of the checkpoint that has been completed.
+	 */
+	void commitCheckpoint(long checkpointId);
+}

flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/checkpoint/Checkpointed.java

+/*
+ * 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.checkpoint;
+
+import org.apache.flink.runtime.state.OperatorState;
+
+/**
+ * This method must be implemented by functions that have state that needs to be
+ * checkpointed. The functions get a call whenever a checkpoint should take place
+ * and return a snapshot of their state, which will be checkpointed.
+ * 
+ * <p>This interface marks a function as <i>synchronously</i> checkpointed. While the
+ * state is written, the function is not called, so the function needs not return a
+ * copy of its state, but may return a reference to its state. Functions that can
+ * continue to work and mutate the state, even while the state snapshot is being accessed,
+ * can implement the {@link org.apache.flink.streaming.api.checkpoint.CheckpointedAsynchronously}
+ * interface.</p>
+ */
+public interface Checkpointed {
+
+	/**
+	 * Gets the current operator state as a checkpoint. The state must reflect all operations
+	 * from all prior operations if this function. 
+	 * 
+	 * @param checkpointId The ID of the checkpoint.
+	 * @param checkpointTimestamp The timestamp of the checkpoint, as derived by
+	 *                            System.currentTimeMillis() on the JobManager.
+	 *                            
+	 * @return A snapshot of the operator state.
+	 * 
+	 * @throws Exception Thrown if the creation of the state object failed. This causes the
+	 *                   checkpoint to fail. The system may decide to fail the operation (and trigger
+	 *                   recovery), or to discard this checkpoint attempt and to continue running
+	 *                   and to try again with the next checkpoint attempt.
+	 */
+	OperatorState<?> snapshotState(long checkpointId, long checkpointTimestamp) throws Exception;
+}

flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/checkpoint/CheckpointedAsynchronously.java

+/*
+ * 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.checkpoint;
+
+/**
+ * This interface marks a function/operator as <i>asynchronously checkpointed</i>.
+ * Similar to the {@link Checkpointed} interface, the function must produce a
+ * snapshot of its state. However, the function must be able to continue working
+ * and mutating its state without mutating the returned state snapshot.
+ * 
+ * <p>Asynchronous checkpoints are desirable, because they allow the data streams at the
+ * point of the checkpointed function/operator to continue running while the checkpoint
+ * is in progress.</p>
+ * 
+ * <p>To be able to support asynchronous snapshots, the state returned by the
+ * {@link #snapshotState(long, long)} method is typically a copy or shadow copy
+ * of the actual state.</p>
+ */
+public interface CheckpointedAsynchronously extends Checkpointed {}

flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/StreamSource.java

+/*
+ * 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.functions.source;
+
+/**
+ * Base interface for all stream data sources in Flink. The contract of a stream source
+ * is similar to an iterator - it is consumed as in the following pseudo code:
+ * 
+ * <pre>{@code
+ * StreamSource<T> source = ...;
+ * Collector<T> out = ...;
+ * while (!source.reachedEnd()) {
+ *   out.collect(source.next());
+ * }
+ * }
+ * </pre>
+ * 
+ * <b>Note about blocking behavior</b>
+ * <p>This implementations of the methods in the stream sources must have certain guarantees about
+ * blocking behavior. One of the two characteristics must be fulfilled.</p>
+ * <ul>
+ *     <li>The methods must react to thread interrupt calls and break out of blocking calls with
+ *         an {@link InterruptedException}.</li>
+ *     <li>The method may ignore interrupt calls and/or swallow InterruptedExceptions, if it is guaranteed
+ *         that the method returns quasi immediately irrespectively of the input. This is true for example
+ *         for file streams, where the call is guaranteed to return after a very short I/O delay in
+ *         the order of milliseconds.</li>
+ * </ul>
+ * 
+ * @param <T> The type of the records produced by this source.
+ */
+public interface StreamSource<T> {
+	
+	/**
+	 * Checks whether the stream has reached its end.
+	 *
+	 * <p>This method must obey the contract about blocking behavior declared in the
+	 * description of this class.</p>
+	 * 
+	 * @return True, if the end of the stream has been reached, false if more data is available.
+	 * 
+	 * @throws InterruptedException The calling thread may be interrupted to pull the function out of this
+	 *                              method during checkpoints.
+	 * @throws Exception Any other exception that is thrown causes the source to fail and results in failure of
+	 *                   the streaming program, or triggers recovery, depending on the program setup.
+	 */
+	boolean reachedEnd() throws Exception;
+
+
+	/**
+	 * Produces the next record.
+	 * 
+	 * <p>This method must obey the contract about blocking behavior declared in the
+	 * description of this class.</p>
+	 * 
+	 * @return The next record produced by this stream source.
+	 * 
+	 * @throws InterruptedException The calling thread may be interrupted to pull the function out of this
+	 *                              method during checkpoints.
+	 * @throws Exception Any other exception that is thrown causes the source to fail and results in failure of
+	 *                   the streaming program, or triggers recovery, depending on the program setup.
+	 */
+	T next() throws Exception;
+}