headers = new HashMap<>();
+ for (KafkaHeader header : record.headers()) {
+ headers.put(header.key(), header.value().toString());
+ }
+ return headers;
+ }
+
+ static void addProperty(Properties props, String key, String value) {
+ String previous = props.getProperty(key);
+ if (previous != null) {
+ props.setProperty(key, previous + "," + value);
+ } else {
+ props.setProperty(key, value);
+ }
+ }
+}
diff --git a/src/test/java/io/strimzi/kafka/bridge/tracing/OpenTelemetryTest.java b/src/test/java/io/strimzi/kafka/bridge/tracing/OpenTelemetryTest.java
new file mode 100644
index 000000000..9956288a6
--- /dev/null
+++ b/src/test/java/io/strimzi/kafka/bridge/tracing/OpenTelemetryTest.java
@@ -0,0 +1,26 @@
+/*
+ * Copyright Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+
+package io.strimzi.kafka.bridge.tracing;
+
+import io.vertx.core.tracing.TracingOptions;
+import io.vertx.tracing.opentelemetry.OpenTelemetryOptions;
+
+import static io.strimzi.kafka.bridge.tracing.TracingConstants.JAEGER;
+import static io.strimzi.kafka.bridge.tracing.TracingConstants.OPENTELEMETRY_SERVICE_NAME_PROPERTY_KEY;
+import static io.strimzi.kafka.bridge.tracing.TracingConstants.OPENTELEMETRY_TRACES_EXPORTER_PROPERTY_KEY;
+
+/**
+ * OpenTelemetry tests
+ */
+public class OpenTelemetryTest extends TracingTestBase {
+ @Override
+ protected TracingOptions tracingOptions() {
+ System.setProperty(OPENTELEMETRY_TRACES_EXPORTER_PROPERTY_KEY, JAEGER);
+ System.setProperty(OPENTELEMETRY_SERVICE_NAME_PROPERTY_KEY, "strimzi-kafka-bridge-test");
+ System.setProperty("otel.metrics.exporter", "none"); // disable metrics
+ return new OpenTelemetryOptions();
+ }
+}
diff --git a/src/test/java/io/strimzi/kafka/bridge/tracing/OpenTracingTest.java b/src/test/java/io/strimzi/kafka/bridge/tracing/OpenTracingTest.java
new file mode 100644
index 000000000..584fa6364
--- /dev/null
+++ b/src/test/java/io/strimzi/kafka/bridge/tracing/OpenTracingTest.java
@@ -0,0 +1,25 @@
+/*
+ * Copyright Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+
+package io.strimzi.kafka.bridge.tracing;
+
+import io.vertx.core.tracing.TracingOptions;
+import io.vertx.tracing.opentracing.OpenTracingOptions;
+
+/**
+ * OpenTracing tests
+ *
+ * These env vars need to be set:
+ * * JAEGER_SERVICE_NAME=ot_kafka_bridge_test
+ * * JAEGER_SAMPLER_TYPE=const
+ * * JAEGER_SAMPLER_PARAM=1
+ */
+public class OpenTracingTest extends TracingTestBase {
+ @Override
+ protected TracingOptions tracingOptions() {
+ System.setProperty("otel.metrics.exporter", "none"); // disable OTel metrics -- they slip in via ServiceLoader
+ return new OpenTracingOptions();
+ }
+}
diff --git a/src/test/java/io/strimzi/kafka/bridge/tracing/TracingTestBase.java b/src/test/java/io/strimzi/kafka/bridge/tracing/TracingTestBase.java
new file mode 100644
index 000000000..179344ef1
--- /dev/null
+++ b/src/test/java/io/strimzi/kafka/bridge/tracing/TracingTestBase.java
@@ -0,0 +1,141 @@
+/*
+ * Copyright Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+
+package io.strimzi.kafka.bridge.tracing;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.strimzi.kafka.bridge.BridgeContentType;
+import io.strimzi.kafka.bridge.http.services.ConsumerService;
+import io.strimzi.kafka.bridge.http.services.ProducerService;
+import io.strimzi.kafka.bridge.utils.Urls;
+import io.vertx.core.AsyncResult;
+import io.vertx.core.Handler;
+import io.vertx.core.Vertx;
+import io.vertx.core.VertxOptions;
+import io.vertx.core.json.JsonArray;
+import io.vertx.core.json.JsonObject;
+import io.vertx.core.tracing.TracingOptions;
+import io.vertx.core.tracing.TracingPolicy;
+import io.vertx.ext.web.client.HttpResponse;
+import io.vertx.ext.web.client.WebClient;
+import io.vertx.ext.web.client.WebClientOptions;
+import io.vertx.ext.web.codec.BodyCodec;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+import org.hamcrest.CoreMatchers;
+import org.junit.jupiter.api.Assumptions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.URL;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+/**
+ * Base for OpenTracing and OpenTelemetry (manual) tests.
+ *
+ * Test will only run if the bridge AND tracing server are up-n-running.
+ */
+@ExtendWith(VertxExtension.class)
+public abstract class TracingTestBase {
+ Logger log = LoggerFactory.getLogger(getClass());
+
+ private void assumeServer(String url) {
+ try {
+ new URL(url).openConnection().getInputStream();
+ } catch (Exception e) {
+ log.info("Cannot connect to server", e);
+ Assumptions.assumeTrue(false, "Server is not running: " + url);
+ }
+ }
+
+ Handler>> verifyOK(VertxTestContext context) {
+ return ar -> {
+ context.verify(() -> {
+ assertThat(ar.succeeded(), is(true));
+ HttpResponse response = ar.result();
+ assertThat(response.statusCode(), is(HttpResponseStatus.OK.code()));
+ });
+ context.completeNow();
+ };
+ }
+
+ @BeforeEach
+ public void setUp() {
+ assumeServer(String.format("http://%s:%s", Urls.BRIDGE_HOST, Urls.BRIDGE_PORT)); // bridge
+ assumeServer("http://localhost:16686"); // jaeger
+ }
+
+ protected abstract TracingOptions tracingOptions();
+
+ @Test
+ public void testSmoke(VertxTestContext context) throws Exception {
+ Vertx vertx = Vertx.vertx(new VertxOptions().setTracingOptions(tracingOptions()));
+
+ WebClient client = WebClient.create(vertx, new WebClientOptions()
+ .setDefaultHost(Urls.BRIDGE_HOST)
+ .setDefaultPort(Urls.BRIDGE_PORT)
+ .setTracingPolicy(TracingPolicy.ALWAYS)
+ );
+
+ String value = "message-value";
+
+ JsonArray records = new JsonArray();
+ JsonObject json = new JsonObject();
+ json.put("value", value);
+ records.add(json);
+
+ JsonObject root = new JsonObject();
+ root.put("records", records);
+
+ String topicName = "mytopic";
+
+ ProducerService.getInstance(client)
+ .sendRecordsRequest(topicName, root, BridgeContentType.KAFKA_JSON_JSON)
+ .sendJsonObject(root, verifyOK(context));
+
+ ConsumerService consumerService = ConsumerService.getInstance(client);
+
+ // create consumer
+ // subscribe to a topic
+
+ String consumerName = "my-consumer";
+ String groupId = UUID.randomUUID().toString();
+
+ JsonObject consumerJson = new JsonObject()
+ .put("name", consumerName)
+ .put("format", "json");
+
+ consumerService
+ .createConsumer(context, groupId, consumerJson)
+ .subscribeConsumer(context, groupId, consumerName, topicName);
+
+ CompletableFuture consume = new CompletableFuture<>();
+ // consume records
+ consumerService
+ .consumeRecordsRequest(groupId, consumerName, BridgeContentType.KAFKA_JSON_JSON)
+ .as(BodyCodec.jsonArray())
+ .send(ar -> {
+ context.verify(() -> {
+ assertThat(ar.succeeded(), CoreMatchers.is(true));
+ HttpResponse response = ar.result();
+ assertThat(response.statusCode(), CoreMatchers.is(HttpResponseStatus.OK.code()));
+ });
+ consume.complete(true);
+ });
+
+ consume.get(60, TimeUnit.SECONDS);
+
+ // consumer deletion
+ consumerService.deleteConsumer(context, groupId, consumerName);
+ }
+}