[camel] 01/02: CAMEL-16815: camel-tracing - Kafka decorator should not cause type cast exception when using avro types.

[da...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git

commit 24251cc83ad8a25925636982d2f18e3159355388
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Fri Jul 30 12:01:42 2021 +0200

    CAMEL-16815: camel-tracing - Kafka decorator should not cause type cast exception when using avro types.
---
 .../camel/tracing/decorators/KafkaSpanDecorator.java      | 15 +++++++--------
 .../camel/tracing/decorators/KafkaSpanDecoratorTest.java  | 10 +++++-----
 2 files changed, 12 insertions(+), 13 deletions(-)

diff --git a/components/camel-tracing/src/main/java/org/apache/camel/tracing/decorators/KafkaSpanDecorator.java b/components/camel-tracing/src/main/java/org/apache/camel/tracing/decorators/KafkaSpanDecorator.java
index 44f111d..b28b857 100644
--- a/components/camel-tracing/src/main/java/org/apache/camel/tracing/decorators/KafkaSpanDecorator.java
+++ b/components/camel-tracing/src/main/java/org/apache/camel/tracing/decorators/KafkaSpanDecorator.java
@@ -50,7 +50,7 @@ public class KafkaSpanDecorator extends AbstractMessagingSpanDecorator {
 
     @Override
     public String getDestination(Exchange exchange, Endpoint endpoint) {
-        String topic = (String) exchange.getIn().getHeader(TOPIC);
+        String topic = exchange.getIn().getHeader(TOPIC, String.class);
         if (topic == null) {
             Map<String, String> queryParameters = toQueryParameters(endpoint.getEndpointUri());
             topic = queryParameters.get("topic");
@@ -67,17 +67,17 @@ public class KafkaSpanDecorator extends AbstractMessagingSpanDecorator {
             span.setTag(KAFKA_PARTITION_TAG, partition);
         }
 
-        String partitionKey = (String) exchange.getIn().getHeader(PARTITION_KEY);
+        String partitionKey = exchange.getIn().getHeader(PARTITION_KEY, String.class);
         if (partitionKey != null) {
             span.setTag(KAFKA_PARTITION_KEY_TAG, partitionKey);
         }
 
-        String key = (String) exchange.getIn().getHeader(KEY);
+        String key = exchange.getIn().getHeader(KEY, String.class);
         if (key != null) {
             span.setTag(KAFKA_KEY_TAG, key);
         }
 
-        String offset = getValue(exchange, OFFSET, Long.class);
+        String offset = getValue(exchange, OFFSET, String.class);
         if (offset != null) {
             span.setTag(KAFKA_OFFSET_TAG, offset);
         }
@@ -86,10 +86,9 @@ public class KafkaSpanDecorator extends AbstractMessagingSpanDecorator {
     /**
      * Extracts header value from the exchange for given header
      * 
-     * @param  exchange the {@link Exchange}
-     * @param  header   the header name
-     * @param  type     the class type of the exchange header
-     * @return
+     * @param exchange the {@link Exchange}
+     * @param header   the header name
+     * @param type     the class type of the exchange header
      */
     private <T> String getValue(final Exchange exchange, final String header, Class<T> type) {
         T partition = exchange.getIn().getHeader(header, type);
diff --git a/components/camel-tracing/src/test/java/org/apache/camel/tracing/decorators/KafkaSpanDecoratorTest.java b/components/camel-tracing/src/test/java/org/apache/camel/tracing/decorators/KafkaSpanDecoratorTest.java
index 24784d6..b88e517 100644
--- a/components/camel-tracing/src/test/java/org/apache/camel/tracing/decorators/KafkaSpanDecoratorTest.java
+++ b/components/camel-tracing/src/test/java/org/apache/camel/tracing/decorators/KafkaSpanDecoratorTest.java
@@ -34,7 +34,7 @@ public class KafkaSpanDecoratorTest {
         Message message = Mockito.mock(Message.class);
 
         Mockito.when(exchange.getIn()).thenReturn(message);
-        Mockito.when(message.getHeader(KafkaSpanDecorator.TOPIC)).thenReturn("test");
+        Mockito.when(message.getHeader(KafkaSpanDecorator.TOPIC, String.class)).thenReturn("test");
 
         KafkaSpanDecorator decorator = new KafkaSpanDecorator();
 
@@ -69,10 +69,10 @@ public class KafkaSpanDecoratorTest {
 
         Mockito.when(endpoint.getEndpointUri()).thenReturn("test");
         Mockito.when(exchange.getIn()).thenReturn(message);
-        Mockito.when(message.getHeader(KafkaSpanDecorator.KEY)).thenReturn(testKey);
+        Mockito.when(message.getHeader(KafkaSpanDecorator.KEY, String.class)).thenReturn(testKey);
         Mockito.when(message.getHeader(KafkaSpanDecorator.OFFSET, String.class)).thenReturn(testOffset);
         Mockito.when(message.getHeader(KafkaSpanDecorator.PARTITION, String.class)).thenReturn(testPartition);
-        Mockito.when(message.getHeader(KafkaSpanDecorator.PARTITION_KEY)).thenReturn(testPartitionKey);
+        Mockito.when(message.getHeader(KafkaSpanDecorator.PARTITION_KEY, String.class)).thenReturn(testPartitionKey);
 
         SpanDecorator decorator = new KafkaSpanDecorator();
 
@@ -97,8 +97,8 @@ public class KafkaSpanDecoratorTest {
 
         Mockito.when(endpoint.getEndpointUri()).thenReturn("test");
         Mockito.when(exchange.getIn()).thenReturn(message);
-        Mockito.when(message.getHeader(KafkaSpanDecorator.OFFSET, Long.class)).thenReturn(testOffset);
-        Mockito.when(message.getHeader(KafkaSpanDecorator.PARTITION, Integer.class)).thenReturn(testPartition);
+        Mockito.when(message.getHeader(KafkaSpanDecorator.OFFSET, String.class)).thenReturn(testOffset.toString());
+        Mockito.when(message.getHeader(KafkaSpanDecorator.PARTITION, String.class)).thenReturn(testPartition.toString());
 
         SpanDecorator decorator = new KafkaSpanDecorator();