Hi everyone,
I'm putting together a Flink workflow that needs to merge historic data from a
custom JDBC source with a Kafka flow (for the realtime data). I have
successfully written the custom JDBC source that emits a watermark for the last
event time after all the DB data has been emitted but now I face a problem when
joining with data from the Kafka stream.
I register a timer in my KeyedCoProcessFunction joining the DB stream with live
Kafka stream so I can emit all the "batch" data from the DB in one go when
completely read up to the watermark but the timer never fires as the Kafka
stream is empty and therefore doesn't emit a watermark. My Kafka stream will
allowed to be empty since all the data will have been retrieved from the DB
call so I only expect new events to appear over Kafka. Note that if I replace
the Kafka input with a simple env.fromCollection(...) empty list then the timer
triggers fine as Flink seems to detect it doesn't need to wait for any input
from stream 2. So it seems to be something related to the Kafka stream status
that is preventing the watermark from advancing in the KeyedCoProcessFunction.
I have tried configuring the Kafka stream timestamp and watermark strategies to
so the source is marked as idle after 10 seconds but still it seems the
watermark in the join operator combining these 2 streams is not advancing. (See
example code below).
Maybe this is my bad understanding but I thought if an input stream into a
KeyedCoProcessFunction is idle then it wouldn't be considered by the operator
for forwarding the watermark i.e. it would forward the non-idle input stream's
watermark and not do a min(stream1WM, stream2WM). With the below example I
never see the onTimer fire and the only effect the withIdleness() strategy has
is to stop the print statements in onPeriodicEmit() happening after 5 seconds
(env periodic emit is set to the default 200ms so I see 25 rows before it
stops).
The only way I can get my KeyedCoProcessFunction timer to fire is to force an
emit of the watermark I want in the onPeriodicEmit() after x numbers of
attempts to advance an initial watermark i.e. if onPeriodicEmit() is called 100
times and the "latestWatermark" is still Long.MIN_VALUE then I emit the
watermark I want so the join can progress. This seems like a nasty hack to me
but perhaps something like this is actually necessary?
I am currently using Flink 1.12.3, a Confluent Kafka client 6.1.1. Any pointers
would be appreciated.
Thanks in advance,
James.
FlinkKafkaConsumer<Position> positionsFlinkKafkaConsumer = new
FlinkKafkaConsumer<>("poc.positions",
ConfluentRegistryAvroDeserializationSchema.forSpecific(Position.class,
SchemaRegistryURL), kafkaConsumerProperties);
positionsFlinkKafkaConsumer.setStartFromEarliest();
positionsFlinkKafkaConsumer.assignTimestampsAndWatermarks(
new WatermarkStrategy<Position>() {
@Override
public TimestampAssigner<Position>
createTimestampAssigner(TimestampAssignerSupplier.Context context) {
return (event, recordTimestamp) -> {
return event.getPhysicalFrom();
};
}
@Override
public WatermarkGenerator<Position>
createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
return new WatermarkGenerator<Position>() {
public long latestWatermark = Long.MIN_VALUE;
@Override
public void onEvent(Position event, long timestamp,
WatermarkOutput output) {
long eventWatermark = event.getPhysicalFrom();
if (eventWatermark > latestWatermark)
latestWatermark = eventWatermark;
}
@Override
public void onPeriodicEmit(WatermarkOutput output) {
System.out.printf("Emitting watermark %d\n",
latestWatermark);
output.emitWatermark(new
Watermark(latestWatermark));
}
};
}
}.withIdleness(Duration.ofSeconds(5)));
DataStream<Position> positionKafkaInputStream =
env.addSource(positionsFlinkKafkaConsumer, "Kafka-Source");
DataStream<Position> otherPositionStream =
env.fromCollection(Lists.newArrayList(new Position(...,
timestamp.getMillis())), TypeInformation.of(Position.class));
otherPositionStream.assignTimestampsAndWatermarks(
WatermarkStrategy
.<Position>forBoundedOutOfOrderness(Duration.ofSeconds(10))
.withTimestampAssigner((e, t) -> e.getPhysicalFrom()));
KeyedStream<Position, String> keyedPositionKafkaInputStream =
positionKafkaInputStream.keyBy(p -> p.getMarketName());
KeyedStream<Position, String> keyedOtherPositionStream =
otherPositionStream.keyBy(p -> p.getMarketName());
DataStream<Position> joinedStream = keyedOtherPositionStream
.connect(keyedPositionKafkaInputStream)
.process(new JoinProcessFunction());
...
private static class JoinProcessFunction extends KeyedCoProcessFunction<String,
Position, Position, Position> {
private static Logger logger =
LoggerFactory.getLogger(JoinProcessFunction.class);
@Override
public void processElement1(Position position, Context context,
Collector<Position> collector) {
logger.info("processPosition1: key: {}, ts: {}, watermark: {}",
context.getCurrentKey(), context.timestamp(),
context.timerService().currentWatermark());
context.timerService().registerEventTimeTimer(position.getPhysicalFrom());
collector.collect(position);
}
@Override
public void processElement2(Position position, Context context,
Collector<Position> collector) {
logger.info("processPosition2: key: {}, ts: {}, watermark: {}",
context.getCurrentKey(), context.timestamp(),
context.timerService().currentWatermark());
context.timerService().registerEventTimeTimer(position.getPhysicalFrom());
collector.collect(position);
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx,
Collector<Position> out) {
logger.info("Timer triggered for timestamp {} and key '{}' -
current Watermark: {}.", timestamp, ctx.getCurrentKey(),
ctx.timerService().currentWatermark());
}
}
