Я пытаюсь интегрировать трассировку Jaeger в K-Streams.Я планировал добавить трассировку к нескольким из моих наиболее важных конвейеров и задавался вопросом, что будет хорошим способом передачи traceid с одной круговой линии на другую?
Вот что у меня есть - в начале потокаобрабатывая конвейер, я запускаю серверный промежуток и сохраняю traceid в хранилище состояний.Позже в конвейере преобразования я получаю доступ к хранилищу состояний и получаю трассировку из метода transform ().Это хороший способ обработки трассировки в потоковой обработке?
input
.mapValues(toSomethingThatMyAppUnderstands)
.mapValues(this::startStreamTrace)
.filter((k, v) -> v.isPresent())
.mapValues(Optional::get)
.mapValues(doSomethingHereWith)
.flatMapValues(doSomethingElse)
.filter((k, v) -> isInterestingEvent(v))
.transform(() -> new TransformerWithTracing<SomeObjectA, SomeObjectB>(IN_MEM_STORE_NAME, someFunction), IN_MEM_STORE_NAME)
.flatMapValues(c -> c)
.to(outTopic, Produced.with(Serdes.String(), new EventSerde()));
public class TransformerWithTracing<V, VR> implements Transformer<String, V, KeyValue<String, VR>> {
final Function valueAction;
final String storeId;
private KeyValueStore<String, String> traceIdStore;
public TransformerWithTracing(String storeId, Function valueAction) {
this.storeId = storeId;
this.valueAction = valueAction;
}
@Override
public void init(ProcessorContext context) {
// KeyValueStore store = ((KeyValueStore<String, String>) context.getStateStore(storeId));
InMemoryKeyValueStore inMemoryKeyValueStore = (InMemoryKeyValueStore) store;
this.traceIdStore = store;
}
@Override
public KeyValue<String, VR> transform(String key, V value) {
System.out.println(traceIdStore.get(key));
// BuildTraceHeader
try(Scope scope = serviceTracer.startServerSpan(traceHeader, "Converting to Enterprise Event")) {
return KeyValue.pair(key, (VR) valueAction.apply(value));
}
}
@Override
public KeyValue<String, VR> punctuate(long timestamp) {
return null;
}
@Override
public void close() {
// if (streamId != null) traceIdStore.delete(streamId);
}
}