@Override
    public void pre(SpanAdapter span, Exchange exchange, Endpoint endpoint) {
        super.pre(span, exchange, endpoint);

        String partition = getValue(exchange, PARTITION, Integer.class);
        if (partition != null) {
            span.setTag(KAFKA_PARTITION_TAG, partition);
        }

        String partitionKey = exchange.getIn().getHeader(PARTITION_KEY, String.class);
        if (partitionKey != null) {
            span.setTag(KAFKA_PARTITION_KEY_TAG, partitionKey);
        }

        String key = exchange.getIn().getHeader(KEY, String.class);
        if (key != null) {
            span.setTag(KAFKA_KEY_TAG, key);
        }

        String offset = getValue(exchange, OFFSET, String.class);
        if (offset != null) {
            span.setTag(KAFKA_OFFSET_TAG, offset);
        }
    }
@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, String.class)).thenReturn(testOffset.toString());
        Mockito.when(message.getHeader(KafkaSpanDecorator.PARTITION, String.class)).thenReturn(testPartition.toString());

        SpanDecorator decorator = new KafkaSpanDecorator();

        MockSpanAdapter span = new MockSpanAdapter();

        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));
    }