/**
 * Hybrid source that switches underlying sources based on configured source chain.
 *
 * <p>A simple example with FileSource and KafkaSource with fixed Kafka start position:
 *
 * <pre>{@code
 * FileSource<String> fileSource =
 *   FileSource.forRecordStreamFormat(new TextLineFormat(), Path.fromLocalFile(testDir)).build();
 * KafkaSource<String> kafkaSource =/**
 * Hybrid source that switches underlying sources based on configured source chainKafkaSource.<String>builder()
 * *                   .setBootstrapServers("localhost:9092")
 * <p>A simple example with FileSource and KafkaSource with fixed Kafka start position:
 *
 * <pre>{@code
 * FileSource<String> fileSource = .setGroupId("MyGroup")
 *   FileSource.forRecordStreamFormat(new TextLineFormat(), Path.fromLocalFile(testDir)).build();
 * KafkaSource<String> kafkaSource =
 *           KafkaSource.<String>builder(.setTopics(Arrays.asList("quickstart-events"))
 *                   .setBootstrapServers("localhost:9092")setDeserializer(
 *                     .setGroupId("MyGroup"      KafkaRecordDeserializer.valueOnly(StringDeserializer.class))
 *                   .setTopicssetStartingOffsets(ArraysOffsetsInitializer.asListearliest("quickstart-events"))
 *                   .setDeserializer(build();
 * HybridSource<String> hybridSource =
 *           HybridSource.builder(fileSource)
 *                    KafkaRecordDeserializer.valueOnly(StringDeserializer.class)addSource(kafkaSource)
 *                   .setStartingOffsetsbuild(OffsetsInitializer.earliest());
 * }</pre>
 *
 * <p>A more complex example with Kafka start position derived from previous source:
 *
   .build();* <pre>{@code
 * HybridSource<String> hybridSource ==
 *     HybridSource.<String, StaticFileSplitEnumerator>builder(fileSource)
 *           HybridSource.builderaddSource(fileSource)
 *             switchContext -> {
 *   .addSource(kafkaSource)
 *           StaticFileSplitEnumerator previousEnumerator =
 *     .build();
 * }</pre>
 *
 * <p>A more complex example with Kafka start position derived from previous source: switchContext.getPreviousEnumerator();
 *
 * <pre>{@code
 * HybridSource<String> hybridSource =
 *     HybridSource.<String, StaticFileSplitEnumerator>builder(fileSource)
 * // how to get timestamp depends on specific .addSource(enumerator
 *               long switchContexttimestamp -> {= previousEnumerator.getEndTimestamp();
 *               StaticFileSplitEnumeratorOffsetsInitializer previousEnumeratoroffsets =
 *                   switchContextOffsetsInitializer.getPreviousEnumeratortimestamp(timestamp);
 *               // how to get timestamp depends on specific enumeratorKafkaSource<String> kafkaSource =
 *               long  timestamp = previousEnumeratorKafkaSource.getEndTimestamp<String>builder();
 *               OffsetsInitializer offsets =        .setBootstrapServers("localhost:9092")
 *                      OffsetsInitializer .timestampsetGroupId(timestamp"MyGroup");
 *                        KafkaSource<String> kafkaSource =
 *.setTopics(Arrays.asList("quickstart-events"))
 *                       KafkaSource.<String>buildersetDeserializer()
 *                           KafkaRecordDeserializer.setBootstrapServers("localhost:9092"valueOnly(StringDeserializer.class))
 *                       .setGroupIdsetStartingOffsets("MyGroup"offsets)
 *                       .setTopics(Arrays.asList("quickstart-events"))build();
 *                       .setDeserializer(return kafkaSource;
 *             },
 *             KafkaRecordDeserializer.valueOnly(StringDeserializer.class))Boundedness.CONTINUOUS_UNBOUNDED)
 *         .build();
 * }</pre>
 */
@PublicEvolving
public class HybridSource<T> implements Source<T, HybridSourceSplit,      .setStartingOffsets(offsets)HybridSourceEnumeratorState> {...}

/**
 * Factory for underlying sources of {@link HybridSource}.
 *
 * <p>This factory permits building of a source at graph construction time or deferred   .build();at switch
 * time. Provides the ability to set a start position in any way a specific returnsource kafkaSource;allows.
 * Future convenience could be built on top of it, for example a default implementation },
 *             Boundedness.CONTINUOUS_UNBOUNDED)
 *         .build();
 * }</pre>
 */
@PublicEvolving
public class HybridSource<T> implements Source<T, HybridSourceSplit, HybridSourceEnumeratorState> {...}

/**
 * Factory for underlying sources of {@link HybridSource}that
 * recognizes optional interfaces to transfer position in a universal format.
 *
 * <p>Called when the current enumerator has finished. The previous source's final state can
 * thus be used to construct the next source, as required for dynamic position transfer at time
 * of switching.
 *
 * <p>If <p>Thisstart factoryposition permitsis buildingknown ofat ajob sourcesubmission attime, graphthe constructionsource timecan orbe deferredconstructed atin switchthe
 * time.entry Providespoint theand abilitysimply towrapped setinto athe startfactory, positionproviding inthe anybenefit wayof avalidation specificduring
 source* allowssubmission.
 * Future convenience could be built on top of it, for example a default implementation that
 * recognizes optional interfaces to transfer position in a universal format.
 *
 * <p>Called when the current enumerator has finished. The previous source's final state can
 * thus be used to construct the next source, as required for dynamic position transfer at time
 * of switching.
 *
 * <p>If start position is known at job submission time, the source can be constructed in the
 * entry point and simply wrapped into the factory, providing the benefit of validation during
 * submission.
 */
@FunctionalInterface
public interface SourceFactory<
                T, SourceT extends Source<T, ?, ?>, FromEnumT extends SplitEnumerator>
        extends Serializable {
    SourceT create(SourceSwitchContext<FromEnumT> context/
@FunctionalInterface
public interface SourceFactory<
                T, SourceT extends Source<T, ?, ?>, FromEnumT extends SplitEnumerator>
        extends Serializable {
    SourceT create(SourceSwitchContext<FromEnumT> context);
}

/**
 * Context provided to source factory.
 *
 * <p>To derive a start position at switch time, the source can be initialized from context of
 * the previous enumerator. A specific enumerator implementation may carry state such as an end
 * timestamp, that can be used to derive the start position of the next source.
 *
 * <p>Currently only the previous enumerator is exposed. The context interface allows for
 * backward compatible extension, i.e. additional information about the previous source can be
 * supplied in the future.
 */
public interface SourceSwitchContext<EnumT> {
    EnumT getPreviousEnumerator();
}