Repository: pinpoint-apm/pinpoint
Issue: Suggest adding RetryingMessageListenerAdapter to KafkaPlugin

**Which module is your feature request related to?**
pinpoint-kafka-plugin
KafkaPlugin.class

**Is your feature request related to a problem?**
I'm using spring-cloud-starter-stream-kafka.
It is set as maxAttempts 3 by default and in this case, it was confirmed to use **RetryingMessageListenerAdapter** as shown below.

![RetryingMessageListenerAdapter](https://user-images.githubusercontent.com/34761230/63135907-9ac02380-c00a-11e9-8952-9cc9e2c780f0.gif)


call stack
`
onMessage:110, RetryingMessageListenerAdapter (org.springframework.kafka.listener.adapter)
onMessage:40, RetryingMessageListenerAdapter (org.springframework.kafka.listener.adapter)
doInvokeOnMessage:1220, KafkaMessageListenerContainer$ListenerConsumer (org.springframework.kafka.listener)
invokeOnMessage:1213, KafkaMessageListenerContainer$ListenerConsumer (org.springframework.kafka.listener)
doInvokeRecordListener:1174, KafkaMessageListenerContainer$ListenerConsumer (org.springframework.kafka.listener)
doInvokeWithRecords:1155, KafkaMessageListenerContainer$ListenerConsumer (org.springframework.kafka.listener)
invokeRecordListener:1096, KafkaMessageListenerContainer$ListenerConsumer (org.springframework.kafka.listener)
invokeListener:924, KafkaMessageListenerContainer$ListenerConsumer (org.springframework.kafka.listener)
pollAndInvoke:740, KafkaMessageListenerContainer$ListenerConsumer (org.springframework.kafka.listener)
run:689, KafkaMessageListenerContainer$ListenerConsumer (org.springframework.kafka.listener)
call:511, Executors$RunnableAdapter (java.util.concurrent)
run:266, FutureTask (java.util.concurrent)
run:748, Thread (java.lang)
`

maxAttempts default setting
`spring.cloud.stream.bindings.<channelName>.consumer.maxAttempts: default 3`
https://docs.spring.io/spring-cloud-https://github.com/naver/pinpoint/blob/master/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/KafkaPlugin.java#L72 



if set entrypoint like below, it works
`profiler.kafka.consumer.entryPoint = org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter`

Just add a few lines. It seems to be more convenient to use.


**Describe the solution you'd like**
https://github.com/naver/pinpoint/blob/master/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/KafkaPlugin.java#L72 
In the above line, adding one line as shown below is expected to work because it implements the onMessage method of the same AcknowledgingConsumerAwareMessageListener interface.
`transformTemplate.transform("org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapterTransform", RecordMessagingMessageListenerAdapterTransform.class);`

You can rename RecordMessagingMessageListenerAdapterTransform-> AcknowledgingConsumerAwareMessageListenerTransform to use it in common, or you can create a RetryingMessageListenerAdapterTransform.class separately


**Additional context**
In addition, there are instructions to leave entryPoint empty when using spring-kafka as shown below. I experienced some failures.

> Setting when using spring-kafka (In this case, you can leave profiler.kafka.consumer.entryPoint option to empty.)
> profiler.springkafka.consumer.enable=true
https://github.com/naver/pinpoint/blob/master/plugins/kafka/README.md


I think it change like below
BatchMessagingMessageListenerAdapter, RecordMessagingMessageListenerAdapter are the default implementations,

If you use RetryingMessageListenerAdapter, FilteringMessageListenerAdapter, FilteringBatchMessageListenerAdapter, etc., it is better to set it at the entrypoint.





Issue comments:
hi @gwagdalf 

I make plugins, but sometimes I am not expert, so I might miss out on what real users need. 

I checked your offer. 
I think it's a very good solution. 

Based on the following method. Could you make code and send me pull request. 
https://github.com/spring-projects/spring-kafka/blob/master/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RetryingMessageListenerAdapter.java#L108

I look forward to hearing from you.

thanks. 

This issue resolved via #5885 


Pull request review comments:

Patch that resolved the issue:
diff --git a/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/KafkaPlugin.java b/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/KafkaPlugin.java
index aaff75dc805c..4f224e82662a 100644
--- a/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/KafkaPlugin.java
+++ b/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/KafkaPlugin.java
@@ -45,7 +45,9 @@
 
 import static com.navercorp.pinpoint.common.util.VarArgs.va;
 
-
+/**
+ * @author Harris Gwag (gwagdalf)
+ */
 public class KafkaPlugin implements ProfilerPlugin, TransformTemplateAware {
 
     private final PLogger logger = PLoggerFactory.getLogger(this.getClass());
@@ -68,10 +70,11 @@ public void setup(ProfilerPluginSetupContext context) {
             transformTemplate.transform("org.apache.kafka.clients.consumer.ConsumerRecord", ConsumerRecordTransform.class);
 
             if (config.isSpringConsumerEnable()) {
-                transformTemplate.transform("org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapterTransform", RecordMessagingMessageListenerAdapterTransform.class);
-
+                transformTemplate.transform("org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter", AcknowledgingConsumerAwareMessageListenerTransform.class);
                 transformTemplate.transform("org.springframework.kafka.listener.adapter.BatchMessagingMessageListenerAdapter", BatchMessagingMessageListenerAdapterTransform.class);
 
+                // Spring Cloud Starter Stream Kafka 2.2.x is supported
+                transformTemplate.transform("org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter", AcknowledgingConsumerAwareMessageListenerTransform.class);
             }
 
             if (StringUtils.hasText(config.getKafkaEntryPoint())) {
@@ -168,7 +171,7 @@ public byte[] doInTransform(Instrumentor instrumentor, ClassLoader classLoader,
 
     }
 
-    public static class RecordMessagingMessageListenerAdapterTransform implements TransformCallback {
+    public static class AcknowledgingConsumerAwareMessageListenerTransform implements TransformCallback {
 
         @Override
         public byte[] doInTransform(Instrumentor instrumentor, ClassLoader classLoader, String className, Class<?> classBeingRedefined, ProtectionDomain protectionDomain, byte[] classfileBuffer) throws InstrumentException {


Test patch: