I've a requirement to add track and span id to Flink jobs running in cluster, the request flows something like below 

**User --> Rest API -> Kafka-topic-1 --> FlinkJob-1 --> Kafka-topic-2 --> FlinkJob-2 --> Consumer --> DB**

I'm using Spring boot to create my rest APIs and using Spring Sleuth to add track and span id to generated logs, the track and span id is added when rest API is invoked and when message is put over Kakfa-topic-1 as well but I'm not able to figure out how to add track and span id while consuming message at FlinkJob-1 and FLinkJob-2 since they are out of spring context.

One way is to make track and span Id to kafka message headers and have Kafka Consumer/Producer interceptor to extract and log track and span Id, I tried this but my interceptors are not invoked as Flink APIs use Flink version of Kafka-client.

Couldn't get my custom KafkaDeserializationSchema invoked 

```
public class MyDeserializationSchema implements KafkaDeserializationSchema<String> {

private static final Logger LOGGER = LoggerFactory.getLogger(MyDeserializationSchema.class);

@Override
public TypeInformation<String> getProducedType() {
    System.out.println("************** Invoked 1");
    LOGGER.debug("************** Invoked 1");
    return null;
}

@Override
public boolean isEndOfStream(String nextElement) {
    System.out.println("************** Invoked 2");
    LOGGER.debug("************** Invoked 2");
    return true;
}

@Override
public String deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
    System.out.println("************** Invoked 3");
    LOGGER.debug("************** Invoked 3");
    return record.toString();
}

 }

```

Can someone please suggest me how to achieve same.