Support Kafka 3.7+ (#707)
This commit is contained in:
darkness-2nd
GitHub
parent
2e08217688
commit
744eb1e5d8
|
|
@ -22,7 +22,7 @@ Release Notes.
|
|||
* Improve 4x performance of ContextManagerExtendService.createTraceContext()
|
||||
* Add a plugin that supports the Solon framework.
|
||||
* Fixed issues in the MySQL component where the executeBatch method could result in empty SQL statements .
|
||||
|
||||
* Support kafka-clients-3.7.x intercept
|
||||
|
||||
All issues and pull requests are [here](https://github.com/apache/skywalking/milestone/213?closed=1)
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,40 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*
|
||||
*/
|
||||
|
||||
package org.apache.skywalking.apm.plugin.kafka.define;
|
||||
|
||||
import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
|
||||
import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
|
||||
|
||||
/**
|
||||
* For Kafka 3.7.x change
|
||||
*
|
||||
* <pre>
|
||||
* 1. The method named pollForFetchs was removed from KafkaConsumer to <code>AsyncKafkaConsumer</code> and <code>LegacyKafkaConsumer</code>
|
||||
* 2. Because of the enhance class was changed, so we should create new Instrumentation to intercept the method
|
||||
* </pre>
|
||||
*/
|
||||
public class Kafka37AsyncConsumerInstrumentation extends KafkaConsumerInstrumentation {
|
||||
|
||||
private static final String ENHANCE_CLASS_37_ASYNC = "org.apache.kafka.clients.consumer.internals.AsyncKafkaConsumer";
|
||||
|
||||
@Override
|
||||
protected ClassMatch enhanceClass() {
|
||||
return byName(ENHANCE_CLASS_37_ASYNC);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,40 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*
|
||||
*/
|
||||
|
||||
package org.apache.skywalking.apm.plugin.kafka.define;
|
||||
|
||||
import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
|
||||
import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
|
||||
|
||||
/**
|
||||
* For Kafka 3.7.x change
|
||||
*
|
||||
* <pre>
|
||||
* 1. The method named pollForFetchs was removed from KafkaConsumer to <code>AsyncKafkaConsumer</code> and <code>LegacyKafkaConsumer</code>
|
||||
* 2. Because of the enhance class was changed, so we should create new Instrumentation to intercept the method
|
||||
* </pre>
|
||||
*/
|
||||
public class Kafka37LegacyConsumerInstrumentation extends KafkaConsumerInstrumentation {
|
||||
|
||||
private static final String ENHANCE_CLASS_37_LEGACY = "org.apache.kafka.clients.consumer.internals.LegacyKafkaConsumer";
|
||||
|
||||
@Override
|
||||
protected ClassMatch enhanceClass() {
|
||||
return byName(ENHANCE_CLASS_37_LEGACY);
|
||||
}
|
||||
}
|
||||
|
|
@ -18,4 +18,6 @@ kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.CallbackInstr
|
|||
kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.KafkaConsumerInstrumentation
|
||||
kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.KafkaProducerInstrumentation
|
||||
kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.KafkaProducerMapInstrumentation
|
||||
kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.KafkaTemplateCallbackInstrumentation
|
||||
kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.KafkaTemplateCallbackInstrumentation
|
||||
kafka-3.7.x=org.apache.skywalking.apm.plugin.kafka.define.Kafka37AsyncConsumerInstrumentation
|
||||
kafka-3.7.x=org.apache.skywalking.apm.plugin.kafka.define.Kafka37LegacyConsumerInstrumentation
|
||||
|
|
|
|||
|
|
@ -54,6 +54,7 @@
|
|||
- jetty-client-9.x
|
||||
- jetty-server-9.x
|
||||
- kafka-0.11.x/1.x/2.x
|
||||
- kafka-3.7.x
|
||||
- kotlin-coroutine
|
||||
- lettuce-5.x
|
||||
- light4j
|
||||
|
|
|
|||
|
|
@ -76,7 +76,7 @@ metrics based on the tracing data.
|
|||
* MQ
|
||||
* [RocketMQ](https://github.com/apache/rocketmq) 3.x-> 5.x
|
||||
* [RocketMQ-gRPC](http://github.com/apache/rocketmq-clients) 5.x
|
||||
* [Kafka](http://kafka.apache.org) 0.11.0.0 -> 3.2.3
|
||||
* [Kafka](http://kafka.apache.org) 0.11.0.0 -> 3.7.1
|
||||
* [Spring-Kafka](https://github.com/spring-projects/spring-kafka) Spring Kafka Consumer 1.3.x -> 2.3.x (2.0.x and 2.1.x not tested and not recommended by [the official document](https://spring.io/projects/spring-kafka))
|
||||
* [ActiveMQ](https://github.com/apache/activemq) 5.10.0 -> 5.15.4
|
||||
* [RabbitMQ](https://www.rabbitmq.com/) 3.x-> 5.x
|
||||
|
|
|
|||
|
|
@ -24,6 +24,7 @@ import java.util.Properties;
|
|||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.function.Consumer;
|
||||
import java.util.Collection;
|
||||
import java.util.regex.Pattern;
|
||||
import java.util.List;
|
||||
import java.util.ArrayList;
|
||||
|
|
@ -32,10 +33,10 @@ import javax.annotation.PostConstruct;
|
|||
import okhttp3.OkHttpClient;
|
||||
import okhttp3.Request;
|
||||
import okhttp3.Response;
|
||||
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecord;
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecords;
|
||||
import org.apache.kafka.clients.consumer.KafkaConsumer;
|
||||
import org.apache.kafka.clients.consumer.internals.NoOpConsumerRebalanceListener;
|
||||
import org.apache.kafka.clients.producer.Callback;
|
||||
import org.apache.kafka.clients.producer.KafkaProducer;
|
||||
import org.apache.kafka.clients.producer.Producer;
|
||||
|
|
@ -270,7 +271,17 @@ public class CaseController {
|
|||
consumerProperties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
|
||||
consumerProperties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
|
||||
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProperties);
|
||||
consumer.subscribe(topicPattern, new NoOpConsumerRebalanceListener());
|
||||
consumer.subscribe(topicPattern, new ConsumerRebalanceListener() {
|
||||
@Override
|
||||
public void onPartitionsRevoked(Collection<TopicPartition> collection) {
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onPartitionsAssigned(Collection<TopicPartition> collection) {
|
||||
|
||||
}
|
||||
});
|
||||
while (true) {
|
||||
if (pollAndInvoke(consumer)) break;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -28,3 +28,6 @@
|
|||
3.0.2
|
||||
3.1.2
|
||||
3.2.3
|
||||
3.6.0
|
||||
3.7.0
|
||||
3.7.1
|
||||
|
|
|
|||
Loading…
Reference in New Issue