Stream processing skeleton code for tracing

SpaasApplication.java

SingleOutputStreamOperator<Record<TraceContext>> sourceStream =
        getSourceBuilder().fromKafka("traces").withOutputType(TraceContext.class).build();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); 
sourceStream.assignTimestampsAndWatermarks(
         new BoundedOutOfOrdernessTimestampExtractor<Record<TraceContext>>(Time.seconds(30)) {
                    @Override
                    public long extractTimestamp(Record<TraceContext> element) {
                        return element.getAttributes(KeystoneAttributes.class).get().getTimestamp();
                    }
         }
 )
 .map(r -> new TraceContextRecord(r))
 .keyBy((KeySelector<TraceContextRecord, String>) value -> value.getTraceContext().getTraceId())
 .window(GlobalWindows.create())
 .allowedLateness(Time.seconds(TracingTrigger.MAX_TIMEOUT.get()))
 .trigger(createTraceTrigger())
 .process(new ProcessWindowFunction<TraceContextRecord, Record<Map<String, Object>>, String, GlobalWindow>() {
            @Override
            public void process(String s, Context context, Iterable<TraceContextRecord> elements,
                                Collector<Record<Map<String, Object>>> out) {
                out.collect(createMessageLossRecord(s, elements));
            }
 })
.addSink(getSinkBuilder().toKafka("lost_messages").build());