/*
 * 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.api.connector.source;

import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.state.CheckpointListener;
import org.apache.flink.core.io.InputStatus;

import java.util.List;
import java.util.concurrent.CompletableFuture;

/**
 * The interface for a source reader which is responsible for reading the records from
 * the source splits assigned by {@link SplitEnumerator}.
 *
 * @param <T> The type of the record emitted by this source reader.
 * @param <SplitT> The type of the the source splits.
 */
@PublicEvolving
public interface SourceReader<T, SplitT extends SourceSplit>
		extends AutoCloseable, CheckpointListener {

	/**
	 * Start the reader.
	 */
	void start();

	/**
	 * Poll the next available record into the {@link SourceOutput}.
	 *
	 * <p>The implementation must make sure this method is non-blocking.
	 *
	 * <p>Although the implementation can emit multiple records into the given SourceOutput,
	 * it is recommended not doing so. Instead, emit one record into the SourceOutput
	 * and return a {@link InputStatus#MORE_AVAILABLE} to let the caller thread
	 * know there are more records available.
	 *
	 * @return The InputStatus of the SourceReader after the method invocation.
	 */
	InputStatus pollNext(ReaderOutput<T> output) throws Exception;

	/**
	 * Checkpoint on the state of the source.
	 *
	 * @return the state of the source.
	 */
	List<SplitT> snapshotState(long checkpointId);

	/**
	 * @return a future that will be completed once there is a record available to poll.
	 */
	CompletableFuture<Void> isAvailable();

	/**
	 * Adds a list of splits for this reader to read.
	 *
	 * @param splits The splits assigned by the split enumerator.
	 */
	void addSplits(List<SplitT> splits);

	/**
	 * Handle a source event sent by the {@link SplitEnumerator}.
	 *
	 * @param sourceEvent the event sent by the {@link SplitEnumerator}.
	 */
	void handleSourceEvents(SourceEvent sourceEvent);

	/**
	 * We have an empty default implementation here because most source readers do not have
	 * to implement the method.
	 *
	 * @see CheckpointListener#notifyCheckpointComplete(long)
	 */
	@Override
	default void notifyCheckpointComplete(long checkpointId) throws Exception {}
}

@@ -2276,190 +2276,1388 @@
 %09 * 
-@return a future that will be completed once there is a record available to poll.%0A%09 */%0A%09CompletableFuture%3CVoid%3E isAvailable();%0A%0A%09/**%0A%09 * Adds a list of splits for this reader to read
+Returns a future that signals that data is available from the reader.%0A%09 *%0A%09 * %3Cp%3EOnce the future completes, the runtime will keep calling the %7B@link #pollNext(ReaderOutput)%7D%0A%09 * method until that methods returns a status other than %7B@link InputStatus#MORE_AVAILABLE%7D.%0A%09 * After that the, the runtime will again call this method to obtain the next future.%0A%09 * Once that completes, it will again call %7B@link #pollNext(ReaderOutput)%7D and so on.%0A%09 *%0A%09 * %3Cp%3EThe contract is the following: If the reader has data available, then all%0A%09 * futures previously returned by this method must eventually complete. Otherwise%0A%09 * the source might stall indefinitely.%0A%09 *%0A%09 * %3Cp%3EIt is not a problem to have occasional %22false positives%22, meaning to complete%0A%09 * a future even if no data is available. However, one should not use an %22always complete%22%0A%09 * future in cases no data is available, because that will result in busy waiting loops%0A%09 * calling %7B@code pollNext(...)%7D even though no data is available.%0A%09 *%0A%09 * @return a future that will be completed once there is a record available to poll.%0A%09 */%0A%09CompletableFuture%3CVoid%3E isAvailable();%0A%0A%09/**%0A%09 * Adds a list of splits for this reader to read.%0A%09 * This method is called when the enumerator assigns a split via%0A%09 * %7B@link SplitEnumeratorContext#assignSplit(SourceSplit, int)%7D or%0A%09 * %7B@link SplitEnumeratorContext#assignSplits(SplitsAssignment)%7D
 .%0A%09 
@@ -3783,16 +3783,23 @@
 andle a 
+custom 
 source e
@@ -3840,16 +3840,159 @@
 rator%7D.%0A
+%09 * This method is called when the enumerator sends an event via%0A%09 * %7B@link SplitEnumeratorContext#sendEventToSourceReader(int, SourceEvent)%7D.%0A
 %09 *%0A%09 * 

Improve JavaDocs of SourceReader.