Generic/Template Programming in Flink
SourceFunction<T>
@Public
public interface SourceFunction<T> extends Function, Serializable { void run(SourceContext<T> ctx) throws Exception; void cancel(); @Public // Interface might be extended in the future with additional methods.
interface SourceContext<T> {
void collect(T element); @PublicEvolving
void collectWithTimestamp(T element, long timestamp); @PublicEvolving
void emitWatermark(Watermark mark); @PublicEvolving
void markAsTemporarilyIdle();
Object getCheckpointLock();
void close();
}
}
@Override
public void run(SourceContext<Integer> ctx) throws Exception {
while ((start < counter || counter == -1) && isRunning) {
synchronized (ctx.getCheckpointLock()) {
ctx.collect(start);
++start; // loop back to 0
if (start == Integer.MAX_VALUE) {
start = 0;
}
}
Thread.sleep(10L);
}
}
AsyncFunction<IN, OUT>
@PublicEvolving
public interface AsyncFunction<IN, OUT> extends Function, Serializable {
void asyncInvoke(IN input, ResultFuture<OUT> resultFuture) throws Exception;
default void timeout(IN input, ResultFuture<OUT> resultFuture) throws Exception {
resultFuture.completeExceptionally(
new TimeoutException("Async function call has timed out."));
}
}
@Override
public void asyncInvoke(final Integer input, final ResultFuture<String> resultFuture) {
executorService.submit(() -> {
// wait for while to simulate async operation here
long sleep = (long) (ThreadLocalRandom.current().nextFloat() * sleepFactor);
try {
Thread.sleep(sleep); if (ThreadLocalRandom.current().nextFloat() < failRatio) {
resultFuture.completeExceptionally(new Exception("wahahahaha..."));
} else {
resultFuture.complete(
Collections.singletonList("key-" + (input % 10)));
}
} catch (InterruptedException e) {
resultFuture.complete(new ArrayList<>(0));
}
});
}
public static <IN, OUT> SingleOutputStreamOperator<OUT> orderedWait(
DataStream<IN> in,
AsyncFunction<IN, OUT> func,
long timeout,
TimeUnit timeUnit,
int capacity) {
return addOperator(in, func, timeUnit.toMillis(timeout), capacity, OutputMode.ORDERED);
}
FlatMapFunction<T, O>
@Public
@FunctionalInterface
public interface FlatMapFunction<T, O> extends Function, Serializable {
void flatMap(T value, Collector<O> out) throws Exception;
}
FlatMapFunction<String, Tuple2<String, Integer>>() {
private static final long serialVersionUID = -938116068682344455L;
@Override
public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
out.collect(new Tuple2<>(value, 1));
}
}