stankiewicz commented on code in PR #39150:
URL: https://github.com/apache/beam/pull/39150#discussion_r3655626704
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java:
##########
@@ -1079,6 +1094,50 @@ public Read<T> withIdAttribute(String idAttribute) {
return toBuilder().setIdAttribute(idAttribute).build();
}
+ public Read<T> withEnableOpenTelemetryTracing() {
+ return
toBuilder().setEnableOpenTelemetryTracing(true).setNeedsAttributes(true).build();
+ }
+
+ static class OpenTelemetryHeaderConsumer extends DoFn<PubsubMessage,
PubsubMessage> {
+ Context extractSpanContext(PubsubMessage message) {
+ TextMapGetter<PubsubMessage> extractMessageAttributes =
+ new TextMapGetter<PubsubMessage>() {
+ @Override
+ public String get(@Nullable PubsubMessage carrier, String key) {
+ String attribute =
+
Preconditions.checkArgumentNotNull(carrier).getAttribute("googclient_" + key);
+ return attribute == null ? "" : attribute;
+ }
+
+ @Override
+ public Iterable<String> keys(PubsubMessage carrier) {
+ Map<String, String> attributeMap = carrier.getAttributeMap();
+ return attributeMap == null ? ImmutableList.of() :
attributeMap.keySet();
+ }
+ };
+ return W3CTraceContextPropagator.getInstance()
+ .extract(Context.current(), message, extractMessageAttributes);
+ }
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
