<groupId>io.swagger.core.v3</groupId>
<artifactId>swagger-annotations</artifactId>
</dependency>
+ <dependency>
+ <groupId>io.opentelemetry.instrumentation</groupId>
+ <artifactId>opentelemetry-kafka-clients-2.6</artifactId>
+ <version>1.25.0-alpha</version>
+ </dependency>
+ <dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-exporter-otlp</artifactId>
+ <version>1.25.0</version>
+ </dependency>
+ <dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-sdk-extension-autoconfigure</artifactId>
+ <version>1.25.0-alpha</version>
+ </dependency>
</dependencies>
</project>
* Copyright (C) 2017-2021 AT&T Intellectual Property. All rights reserved.
* Modifications Copyright (C) 2018 Samsung Electronics Co., Ltd.
* Modifications Copyright (C) 2020,2023 Bell Canada. All rights reserved.
- * Modifications Copyright (C) 2022-2023 Nordix Foundation.
+ * Modifications Copyright (C) 2022-2024 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
import com.att.nsa.cambria.client.CambriaClientBuilders;
import com.att.nsa.cambria.client.CambriaClientBuilders.ConsumerBuilder;
import com.att.nsa.cambria.client.CambriaConsumer;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanContext;
+import io.opentelemetry.api.trace.TraceFlags;
+import io.opentelemetry.api.trace.TraceState;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.instrumentation.kafkaclients.v2_6.TracingConsumerInterceptor;
import java.io.IOException;
import java.net.MalformedURLException;
+import java.nio.charset.StandardCharsets;
import java.security.GeneralSecurityException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
-import java.util.Map;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import lombok.Data;
import lombok.Getter;
+import lombok.NoArgsConstructor;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.header.Headers;
import org.jetbrains.annotations.NotNull;
import org.onap.dmaap.mr.client.MRClientFactory;
import org.onap.dmaap.mr.client.impl.MRConsumerImpl;
protected KafkaConsumer<String, String> consumer;
protected Properties kafkaProps;
+ protected boolean allowTracing;
+
/**
* Kafka Consumer Wrapper.
* BusTopicParam - object contains the following parameters
if (kafkaProps.get(ConsumerConfig.GROUP_ID_CONFIG) == null) {
kafkaProps.setProperty(ConsumerConfig.GROUP_ID_CONFIG, busTopicParams.getConsumerGroup());
}
+ if (busTopicParams.isAllowTracing()) {
+ this.allowTracing = true;
+ kafkaProps.setProperty(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG,
+ TracingConsumerInterceptor.class.getName());
+ }
+
consumer = new KafkaConsumer<>(kafkaProps);
//Subscribe to the topic
consumer.subscribe(List.of(busTopicParams.getTopic()));
}
List<String> messages = new ArrayList<>(records.count());
try {
+ if (allowTracing) {
+ createParentTraceContext(records);
+ }
+
for (TopicPartition partition : records.partitions()) {
List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
for (ConsumerRecord<String, String> partitionRecord : partitionRecords) {
return messages;
}
+ private void createParentTraceContext(ConsumerRecords<String, String> records) {
+ TraceParentInfo traceParentInfo = new TraceParentInfo();
+ for (ConsumerRecord<String, String> consumerRecord : records) {
+
+ Headers consumerRecordHeaders = consumerRecord.headers();
+ traceParentInfo = processTraceParentHeader(consumerRecordHeaders);
+ }
+
+ SpanContext spanContext = SpanContext.createFromRemoteParent(
+ traceParentInfo.getTraceId(), traceParentInfo.getSpanId(),
+ TraceFlags.getSampled(), TraceState.builder().build());
+
+ Context.current().with(Span.wrap(spanContext)).makeCurrent();
+ }
+
+ private TraceParentInfo processTraceParentHeader(Headers headers) {
+ TraceParentInfo traceParentInfo = new TraceParentInfo();
+ if (headers.lastHeader("traceparent") != null) {
+ traceParentInfo.setParentTraceId(new String(headers.lastHeader(
+ "traceparent").value(), StandardCharsets.UTF_8));
+
+ String[] parts = traceParentInfo.getParentTraceId().split("-");
+ traceParentInfo.setTraceId(parts[1]);
+ traceParentInfo.setSpanId(parts[2]);
+ }
+
+ return traceParentInfo;
+ }
+
+ @Data
+ @NoArgsConstructor
+ private static class TraceParentInfo {
+ private String parentTraceId;
+ private String traceId;
+ private String spanId;
+ }
+
@Override
public void close() {
super.close();
import com.att.nsa.apiClient.http.HttpClient.ConnectionType;
import com.att.nsa.cambria.client.CambriaBatchingPublisher;
import com.att.nsa.cambria.client.CambriaClientBuilders;
+import io.opentelemetry.instrumentation.kafkaclients.v2_6.TracingProducerInterceptor;
import java.net.MalformedURLException;
import java.security.GeneralSecurityException;
import java.util.ArrayList;
kafkaProps.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KEY_SERIALIZER);
}
+ if (busTopicParams.isAllowTracing()) {
+ kafkaProps.setProperty(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,
+ TracingProducerInterceptor.class.getName());
+ }
+
producer = new KafkaProducer<>(kafkaProps);
}
*/
protected boolean useHttps;
+ /**
+ * Allow tracing.
+ */
+ protected boolean allowTracing;
+
/**
* allow self signed certificates.
*/
* apiKey API Key
* apiSecret API Secret
* useHttps does connection use HTTPS?
+ * allowTracing Is tracing allowed?
* allowSelfSignedCerts are self-signed certificates allow
* @param busTopicParams holds all our parameters
* @throws IllegalArgumentException if invalid parameters are present
this.apiKey = busTopicParams.getApiKey();
this.apiSecret = busTopicParams.getApiSecret();
this.useHttps = busTopicParams.isUseHttps();
+ this.allowTracing = busTopicParams.isAllowTracing();
this.allowSelfSignedCerts = busTopicParams.isAllowSelfSignedCerts();
}
* longitude DME2 Longitude
* additionalProps Additional properties to pass to DME2
* useHttps does connection use HTTPS?
+ * allowTracing is message tracing allowed?
* allowSelfSignedCerts are self-signed certificates allow
*/
@Getter
private int fetchTimeout;
private int fetchLimit;
private boolean useHttps;
+ private boolean allowTracing;
private boolean allowSelfSignedCerts;
private boolean managed;
return this;
}
+ public TopicParamsBuilder allowTracing(boolean allowTracing) {
+ this.params.allowTracing = allowTracing;
+ return this;
+ }
+
public TopicParamsBuilder allowSelfSignedCerts(boolean allowSelfSignedCerts) {
this.params.allowSelfSignedCerts = allowSelfSignedCerts;
return this;
* apiSecret api secret
* partitionId partition id
* useHttps does connection use HTTPS?
+ * allowTracing is tracing allowed?
* allowSelfSignedCerts are self-signed certificates allow *
* @throws IllegalArgumentException if invalid parameters are passed in
*/
* longitude DME2 Longitude
* additionalProps Additional properties to pass to DME2
* useHttps does connection use HTTPS?
+ * allowTracing is tracing allowed?
* allowSelfSignedCerts are self-signed certificates allow
* @param busTopicParams Contains the above mentioned parameters
* @throws IllegalArgumentException An invalid parameter passed in
.userName(this.userName)
.password(this.password)
.useHttps(this.useHttps)
+ .allowTracing(this.allowTracing)
.allowSelfSignedCerts(this.allowSelfSignedCerts)
.build());
} else {
.longitude(this.longitude)
.additionalProps(this.additionalProps)
.useHttps(this.useHttps)
+ .allowTracing(this.allowTracing)
.build());
}
.servers(this.servers)
.topic(this.effectiveTopic)
.useHttps(this.useHttps)
+ .allowTracing(this.allowTracing)
.additionalProps(this.additionalProps)
.build());
logger.info("{}: KAFKA SINK created", this);
* apiSecret the api secret (optional)
* partitionId the partition key (optional, autogenerated if not provided)
* useHttps does connection use HTTPS?
+ * allowTracing is tracing allowed?
* allowSelfSignedCerts are self-signed certificates allow
* @param busTopicParams contains attributes needed
* @throws IllegalArgumentException if invalid arguments are detected
.apiKey(this.apiKey)
.apiSecret(this.apiSecret)
.useHttps(this.useHttps)
+ .allowTracing(this.allowTracing)
.allowSelfSignedCerts(this.allowSelfSignedCerts)
.build());
logger.info("{}: UEB SINK created", this);
.consumerInstance(this.consumerInstance)
.fetchTimeout(this.fetchTimeout)
.fetchLimit(this.fetchLimit)
- .useHttps(this.useHttps);
+ .useHttps(this.useHttps)
+ .allowTracing(this.allowTracing);
if (anyNullOrEmpty(this.userName, this.password)) {
this.consumer = new BusConsumer.CambriaConsumerWrapper(builder
.topic(this.effectiveTopic)
.fetchTimeout(this.fetchTimeout)
.consumerGroup(this.consumerGroup)
- .useHttps(this.useHttps);
+ .useHttps(this.useHttps)
+ .allowTracing(this.allowTracing);
this.consumer = new BusConsumer.KafkaConsumerWrapper(builder
.additionalProps(this.additionalProps)
.fetchTimeout(this.fetchTimeout)
.fetchLimit(this.fetchLimit)
.useHttps(this.useHttps)
+ .allowTracing(this.allowTracing)
.allowSelfSignedCerts(this.allowSelfSignedCerts).build());
}
* policy-endpoints
* ================================================================================
* Copyright (C) 2018-2019 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2024 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
.fetchLimit(MY_FETCH_LIMIT).fetchTimeout(MY_FETCH_TIMEOUT).hostname(MY_HOST).latitude(MY_LAT)
.longitude(MY_LONG).managed(true).partitionId(MY_PARTITION).partner(MY_PARTNER)
.password(MY_PASS).port(MY_PORT).servers(servers).topic(MY_TOPIC)
- .effectiveTopic(MY_EFFECTIVE_TOPIC).useHttps(true).userName(MY_USERNAME)
+ .effectiveTopic(MY_EFFECTIVE_TOPIC).useHttps(true).allowTracing(true).userName(MY_USERNAME)
.serializationProvider(MY_SERIALIZER);
}
return BusTopicParams.builder().additionalProps(addProps).basePath(MY_BASE_PATH).clientName(MY_CLIENT_NAME)
.consumerGroup(MY_CONS_GROUP).consumerInstance(MY_CONS_INST).environment(MY_ENV)
- .hostname(MY_HOST).partitionId(MY_PARTITION).partner(MY_PARTNER)
+ .hostname(MY_HOST).partitionId(MY_PARTITION).partner(MY_PARTNER).fetchTimeout(MY_FETCH_TIMEOUT)
.port(KAFKA_PORT).servers(servers).topic(MY_TOPIC)
- .effectiveTopic(MY_EFFECTIVE_TOPIC).useHttps(false);
+ .effectiveTopic(MY_EFFECTIVE_TOPIC).useHttps(false).allowTracing(true);
}
}
* policy-endpoints
* ================================================================================
* Copyright (C) 2018-2021 AT&T Intellectual Property. All rights reserved.
- * Modifications Copyright (C) 2023 Nordix Foundation.
+ * Modifications Copyright (C) 2023-2024 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import com.att.nsa.cambria.client.CambriaConsumer;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import org.apache.commons.collections4.IteratorUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.TopicPartition;
import org.junit.Before;
import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
import org.onap.dmaap.mr.client.impl.MRConsumerImpl;
import org.onap.dmaap.mr.client.response.MRConsumerResponse;
import org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase;
private static final int SHORT_TIMEOUT_MILLIS = 10;
private static final int LONG_TIMEOUT_MILLIS = 3000;
+ @Mock
+ KafkaConsumer<String, String> mockedKafkaConsumer;
+
@Before
@Override
public void setUp() {
super.setUp();
+ MockitoAnnotations.initMocks(this);
}
+
@Test
public void testFetchingBusConsumer() {
// should not be negative
consumer.close();
}
+ @Test
+ public void testFetchNoMessages() throws IOException {
+ KafkaConsumerWrapper kafkaConsumerWrapper = new KafkaConsumerWrapper(makeKafkaBuilder().build());
+ kafkaConsumerWrapper.consumer = mockedKafkaConsumer;
+
+ when(mockedKafkaConsumer.poll(any())).thenReturn(new ConsumerRecords<>(Collections.emptyMap()));
+
+ Iterable<String> result = kafkaConsumerWrapper.fetch();
+
+ verify(mockedKafkaConsumer, times(1)).poll(any());
+
+ assertThat(result != null);
+
+ assertThat(!result.iterator().hasNext());
+
+ mockedKafkaConsumer.close();
+ }
+
+ @Test
+ public void testFetchWithMessages() {
+ // Setup
+ KafkaConsumerWrapper kafkaConsumerWrapper = new KafkaConsumerWrapper(makeKafkaBuilder().build());
+ kafkaConsumerWrapper.consumer = mockedKafkaConsumer;
+
+ ConsumerRecord<String, String> record = new ConsumerRecord<>("my-effective-topic", 0, 0, "key", "value");
+ Map<TopicPartition, List<ConsumerRecord<String, String>>> recordsMap = new HashMap<>();
+ recordsMap.put(new TopicPartition("my-effective-topic", 0), Collections.singletonList(record));
+ ConsumerRecords<String, String> consumerRecords = new ConsumerRecords<>(recordsMap);
+
+ when(mockedKafkaConsumer.poll(any())).thenReturn(consumerRecords);
+
+ Iterable<String> result = kafkaConsumerWrapper.fetch();
+
+ verify(mockedKafkaConsumer, times(1)).poll(any());
+
+ verify(mockedKafkaConsumer, times(1)).commitSync(any(Map.class));
+
+ assertThat(result != null);
+
+ assertThat(result.iterator().hasNext());
+
+ assertThat(result.iterator().next().equals("value"));
+
+ mockedKafkaConsumer.close();
+ }
+
+ @Test
+ public void testFetchWithMessagesAndTraceparent() {
+ // Setup
+ KafkaConsumerWrapper kafkaConsumerWrapper = new KafkaConsumerWrapper(makeKafkaBuilder().build());
+ kafkaConsumerWrapper.consumer = mockedKafkaConsumer;
+
+ ConsumerRecord<String, String> record = new ConsumerRecord<>("my-effective-topic", 0, 0, "key", "value");
+ record.headers().add(
+ "traceparent",
+ "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01".getBytes(StandardCharsets.UTF_8)
+ );
+
+ Map<TopicPartition, List<ConsumerRecord<String, String>>> recordsMap = new HashMap<>();
+ recordsMap.put(new TopicPartition("my-effective-topic", 0), Collections.singletonList(record));
+ ConsumerRecords<String, String> consumerRecords = new ConsumerRecords<>(recordsMap);
+
+ when(mockedKafkaConsumer.poll(any())).thenReturn(consumerRecords);
+
+ Iterable<String> result = kafkaConsumerWrapper.fetch();
+
+ verify(mockedKafkaConsumer, times(1)).poll(any());
+
+ verify(mockedKafkaConsumer, times(1)).commitSync(any(Map.class));
+
+ assertThat(result != null);
+
+ assertThat(result.iterator().hasNext());
+
+ assertThat(result.iterator().next().equals("value"));
+
+ mockedKafkaConsumer.close();
+ }
+
+
@Test
public void testKafkaConsumerWrapperClose() {
assertThatCode(() -> new KafkaConsumerWrapper(makeKafkaBuilder().build()).close()).doesNotThrowAnyException();
"apiKey" : "my-api-key",
"apiSecret" : "my-api-secret",
"useHttps" : true,
+ "allowTracing": false,
"allowSelfSignedCerts" : true,
"consumerGroup" : "${obj.topicSources[0].consumerGroup}",
"consumerInstance" : "${obj.topicSources[0].consumerInstance}",
"apiKey" : "my-api-key",
"apiSecret" : "my-api-secret",
"useHttps" : true,
+ "allowTracing": false,
"allowSelfSignedCerts" : true,
"consumerGroup" : "my-cons-group",
"consumerInstance" : "my-cons-inst",
"apiKey" : "my-api-key",
"apiSecret" : "my-api-secret",
"useHttps" : true,
+ "allowTracing": false,
"allowSelfSignedCerts" : true,
"topicCommInfrastructure" : "UEB",
"partitionKey" : "${obj.topicSinks[0].partitionKey}"
"apiKey" : "my-api-key",
"apiSecret" : "my-api-secret",
"useHttps" : true,
+ "allowTracing": false,
"allowSelfSignedCerts" : true,
"topicCommInfrastructure" : "DMAAP",
"partitionKey" : "my-partition"
"apiKey" : "my-api-key",
"apiSecret" : "my-api-secret",
"useHttps" : true,
+ "allowTracing": true,
"allowSelfSignedCerts" : true,
"topicCommInfrastructure" : "NOOP"
}
"apiKey" : "my-api-key",
"apiSecret" : "my-api-secret",
"useHttps" : true,
+ "allowTracing": true,
"allowSelfSignedCerts" : true,
"topicCommInfrastructure" : "NOOP",
"partitionKey" : "my-partition"
"apiKey" : "my-api-key",
"apiSecret" : "my-api-secret",
"useHttps" : true,
+ "allowTracing": true,
"allowSelfSignedCerts" : true,
"topicCommInfrastructure" : "DMAAP",
"partitionKey" : "my-partition"
"alive": false,
"locked": false,
"useHttps": false,
+ "allowTracing": false,
"topicCommInfrastructure": "KAFKA",
"partitionKey": "my-partition",
"additionalProps": {
"apiKey" : "my-api-key",
"apiSecret" : "my-api-secret",
"useHttps" : true,
+ "allowTracing": true,
"allowSelfSignedCerts" : true,
"topicCommInfrastructure" : "UEB",
"partitionKey" : "my-partition"
"apiKey" : "my-api-key",
"apiSecret" : "my-api-secret",
"useHttps" : true,
+ "allowTracing": true,
"allowSelfSignedCerts" : true,
"consumerGroup" : "my-cons-group",
"consumerInstance" : "my-cons-inst",
"apiKey" : "my-api-key",
"apiSecret" : "my-api-secret",
"useHttps" : true,
+ "allowTracing": true,
"allowSelfSignedCerts" : true,
"consumerGroup" : "my-cons-group",
"consumerInstance" : "my-cons-inst",
"alive": false,
"locked": false,
"useHttps": false,
+ "allowTracing": false,
"topicCommInfrastructure": "KAFKA",
"additionalProps": {
"security.protocol": "SASL_PLAINTEXT",
"apiKey" : "my-api-key",
"apiSecret" : "my-api-secret",
"useHttps" : true,
+ "allowTracing": true,
"allowSelfSignedCerts" : true,
"consumerGroup" : "my-cons-group",
"consumerInstance" : "my-cons-inst",
"apiSecret" : "my-api-secret",
"port": 123,
"useHttps" : true,
+ "allowTracing": true,
"allowSelfSignedCerts" : true,
"consumerGroup" : "consumer group",
"consumerInstance" : "consumer instance",
"apiSecret" : "my-api-secret",
"port": 123,
"useHttps" : true,
+ "allowTracing": true,
"allowSelfSignedCerts" : true,
"consumerGroup" : "consumer group",
"consumerInstance" : "consumer instance",
Code Review / policy / common.git / commitdiff