[Bug] plugins of Pulsar clients for skywalking

[Allen123Hao]

Search before asking

• I searched in the issues and found nothing similar.

Version

OS: k8s pod
Pulsar version:
broker:2.10.2
pulsar-client:3.0.0

Minimal reproduce step

I use pulsar，set consumer config。add negativeAckRedeliveryDelay、deadLetterPolicy params，like this:

       consumer = pulsarClient.newConsumer(Schema.STRING)
                .consumerName("order-payment")
                .topic(TOPIC_PAYMENT)
                .subscriptionName("order-payment-sub")
                .subscriptionType(SubscriptionType.Shared)
                .enableRetry(true)
                .negativeAckRedeliveryDelay(30, TimeUnit.SECONDS)
                .deadLetterPolicy(DeadLetterPolicy.builder()
                        .maxRedeliverCount(3)
                        .retryLetterTopic(RETRY_LETTER_TOPIC_PAYMENT)
                        .deadLetterTopic(DEAD_LETTER_TOPIC_PAYMENT)
                        .build())
                .messageListener(orderConsumer)
                .subscribe();

when send message,the error occur:

java.lang.ClassCastException: class java.lang.String cannot be cast to class org.apache.skywalking.apm.plugin.pulsar.common.MessageEnhanceRequiredInfo (java.lang.String is in module java.base of loader 'bootstrap'; org.apache.skywalking.apm.plugin.pulsar.common.MessageEnhanceRequiredInfo is in unnamed module of loader org.apache.skywalking.apm.agent.core.plugin.loader.AgentClassLoader @2e4262d2)
    at org.apache.skywalking.apm.plugin.pulsar.common.PulsarConsumerListenerInterceptor.lambda$afterMethod$cf935be7$1(PulsarConsumerListenerInterceptor.java:60)
    at org.apache.pulsar.client.impl.ConsumerBase.callMessageListener(ConsumerBase.java:1147)

What did you expect to see?

there is no errors，consumer can consume message normally。

What did you see instead?

the error appearance:

java.lang.ClassCastException: class java.lang.String cannot be cast to class org.apache.skywalking.apm.plugin.pulsar.common.MessageEnhanceRequiredInfo (java.lang.String is in module java.base of loader 'bootstrap'; org.apache.skywalking.apm.plugin.pulsar.common.MessageEnhanceRequiredInfo is in unnamed module of loader org.apache.skywalking.apm.agent.core.plugin.loader.AgentClassLoader @2e4262d2)
    at org.apache.skywalking.apm.plugin.pulsar.common.PulsarConsumerListenerInterceptor.lambda$afterMethod$cf935be7$1(PulsarConsumerListenerInterceptor.java:60)
    at org.apache.pulsar.client.impl.ConsumerBase.callMessageListener(ConsumerBase.java:1147)

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[github-actions]

The issue had no activity for 30 days, mark with Stale label.