添加跟踪和span id到Flink作业

Question

我需要向集群中运行的Flink作业添加track和span id，请求流如下所示

使用者--

我使用Spring Boot来创建我的rest API，并使用Spring Sleuth来添加跟踪和span id到生成的日志中，当调用rest API时添加跟踪和span id，当消息被放在Kakfa-toption-1上时也添加跟踪和span id，但我不能弄清楚如何添加跟踪和跨度ID，同时在Flink作业-1和FLink作业-2上消费消息，因为它们不在Spring上下文中。

一种方法是将track和span Id设置为kafka消息头，并让kafka消费者/生产者拦截器提取和记录track和span Id，我尝试了这个方法，但我的拦截器没有被调用，因为Flink API使用Flink版本的kafka客户机。

无法调用我的自定义KafkaDeserializationSchema

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();
}

 }

有人能建议我如何实现同样的目标吗。

Answer 1

您在这里使用的是一个简单的字符串，在序列化字节到字符串时，可以执行如下代码所示的操作。

public class MyDeserializationSchema  implements KafkaDeserializationSchema<String> {
    @Override
    public boolean isEndOfStream(String nextElement) {
        return false;
    }

    @Override
    public String deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
        return new String(record.value(), "UTF-8");
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }
  }

Answer 2

您也可以使用KafkaDeserializationSchema来获取标题

为了访问Kafka消息的键、值和元数据，KafkaDeserializationSchema具有以下反序列化方法T deserialize（ConsumerRecord记录）。

public class Bla implements KafkaDeserializationSchema {
    @Override
    public boolean isEndOfStream(Object dcEvents) {
        return false;
    }

    @Override
    public Object deserialize(ConsumerRecord consumerRecord) throws Exception {
        return null;
    }



    @Override
    public TypeInformation<DCEvents> getProducedType() {
        return null;
    }