[
https://issues.apache.org/jira/browse/CAMEL-12573?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16510675#comment-16510675
]
ASF GitHub Bot commented on CAMEL-12573:
----------------------------------------
oscerd closed pull request #2372: CAMEL-12573: Fixing class cast exception
URL: https://github.com/apache/camel/pull/2372
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/components/camel-opentracing/src/main/java/org/apache/camel/opentracing/decorators/KafkaSpanDecorator.java
b/components/camel-opentracing/src/main/java/org/apache/camel/opentracing/decorators/KafkaSpanDecorator.java
index ecc92797d58..1ec0a6886eb 100644
---
a/components/camel-opentracing/src/main/java/org/apache/camel/opentracing/decorators/KafkaSpanDecorator.java
+++
b/components/camel-opentracing/src/main/java/org/apache/camel/opentracing/decorators/KafkaSpanDecorator.java
@@ -58,7 +58,7 @@ public String getDestination(Exchange exchange, Endpoint
endpoint) {
public void pre(Span span, Exchange exchange, Endpoint endpoint) {
super.pre(span, exchange, endpoint);
- String partition = (String)exchange.getIn().getHeader(PARTITION);
+ String partition = getValue(exchange, PARTITION, Integer.class);
if (partition != null) {
span.setTag(KAFKA_PARTITION_TAG, partition);
}
@@ -73,10 +73,22 @@ public void pre(Span span, Exchange exchange, Endpoint
endpoint) {
span.setTag(KAFKA_KEY_TAG, key);
}
- String offset = (String)exchange.getIn().getHeader(OFFSET);
+ String offset = getValue(exchange, OFFSET, Long.class);
if (offset != null) {
span.setTag(KAFKA_OFFSET_TAG, offset);
}
}
+ /**
+ * 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
+ */
+ private <T> String getValue(final Exchange exchange, final String header,
Class<T> type) {
+ T partition = exchange.getIn().getHeader(header, type);
+ return partition != null ? String.valueOf(partition) :
exchange.getIn().getHeader(header, String.class);
+ }
+
}
diff --git
a/components/camel-opentracing/src/test/java/org/apache/camel/opentracing/decorators/KafkaSpanDecoratorTest.java
b/components/camel-opentracing/src/test/java/org/apache/camel/opentracing/decorators/KafkaSpanDecoratorTest.java
index b3eda9c3770..2c51def5b47 100644
---
a/components/camel-opentracing/src/test/java/org/apache/camel/opentracing/decorators/KafkaSpanDecoratorTest.java
+++
b/components/camel-opentracing/src/test/java/org/apache/camel/opentracing/decorators/KafkaSpanDecoratorTest.java
@@ -19,6 +19,7 @@
import io.opentracing.mock.MockSpan;
import io.opentracing.mock.MockTracer;
+import jdk.nashorn.internal.IntDeque;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
@@ -59,7 +60,7 @@ public void testGetDestinationNoHeaderTopic() {
}
@Test
- public void testPre() {
+ public void testPreOffsetAndPartitionAsStringHeader() {
String testKey = "TestKey";
String testOffset = "TestOffset";
String testPartition = "TestPartition";
@@ -72,8 +73,8 @@ public void testPre() {
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.OFFSET)).thenReturn(testOffset);
-
Mockito.when(message.getHeader(KafkaSpanDecorator.PARTITION)).thenReturn(testPartition);
+ 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);
SpanDecorator decorator = new KafkaSpanDecorator();
@@ -89,4 +90,29 @@ public void testPre() {
assertEquals(testPartitionKey,
span.tags().get(KafkaSpanDecorator.KAFKA_PARTITION_KEY_TAG));
}
+ @Test
+ public void testPrePartitionAsIntegerHeaderAndOffsetAsLongHeader() {
+ Long testOffset = 4875454l;
+ Integer testPartition = 0;
+
+ Endpoint endpoint = Mockito.mock(Endpoint.class);
+ Exchange exchange = Mockito.mock(Exchange.class);
+ Message message = Mockito.mock(Message.class);
+
+ 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);
+
+ SpanDecorator decorator = new KafkaSpanDecorator();
+
+ MockTracer tracer = new MockTracer();
+ MockSpan span = tracer.buildSpan("TestSpan").start();
+
+ decorator.pre(span, exchange, endpoint);
+
+ assertEquals(String.valueOf(testOffset),
span.tags().get(KafkaSpanDecorator.KAFKA_OFFSET_TAG));
+ assertEquals(String.valueOf(testPartition),
span.tags().get(KafkaSpanDecorator.KAFKA_PARTITION_TAG));
+ }
+
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
> ClassCastException thrown KafkaSpanDecorator
> --------------------------------------------
>
> Key: CAMEL-12573
> URL: https://issues.apache.org/jira/browse/CAMEL-12573
> Project: Camel
> Issue Type: Bug
> Components: camel-opentracing
> Reporter: ayache khettar
> Assignee: Andrea Cosentino
> Priority: Major
> Fix For: 2.20.4, 2.21.2, 2.22.0
>
>
> ClassCastException is thrown by `KafkaSpanDecorator` for the Partition and
> the Offset header values. The fix to this is to cast the partition to an
> Integer and the Offset to a Long. I have got a fix for it and here is a pull
> request for it: https://github.com/apache/camel/pull/2372
> {code:java}
> 2018-06-12 16:36:26.224 WARN 34820 --- [kaConsumer[ocr]]
> o.a.camel.opentracing.OpenTracingTracer : OpenTracing: Failed to capture
> tracing data
> java.lang.ClassCastException: java.lang.Integer cannot be cast to
> java.lang.String
> at
> org.apache.camel.opentracing.decorators.KafkaSpanDecorator.pre(KafkaSpanDecorator.java:61)
> ~[camel-opentracing-2.21.0.jar:2.21.0]
> at
> org.apache.camel.opentracing.OpenTracingTracer$OpenTracingRoutePolicy.onExchangeBegin(OpenTracingTracer.java:297)
> ~[camel-opentracing-2.21.0.jar:2.21.0]
> at
> org.apache.camel.processor.CamelInternalProcessor$RoutePolicyAdvice.before(CamelInternalProcessor.java:495)
> [camel-core-2.21.0.jar:2.21.0]
> at
> org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:149)
> [camel-core-2.21.0.jar:2.21.0]
> at
> org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:97)
> [camel-core-2.21.0.jar:2.21.0]
> at
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:295)
> [camel-kafka-2.21.0.jar:2.21.0]
> at
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:187)
> [camel-kafka-2.21.0.jar:2.21.0]
> at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> [na:1.8.0_151]
> at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_151]
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> [na:1.8.0_151]
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> [na:1.8.0_151]
> at java.lang.Thread.run(Thread.java:748) [na:1.8.0_151]
> {code}
> {code:java}
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
