darcydai commented on a change in pull request #6781:
URL: https://github.com/apache/skywalking/pull/6781#discussion_r616345532
##########
File path:
apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaJVMMetricsSender.java
##########
@@ -38,55 +39,52 @@
* A report to send JVM Metrics data to Kafka Broker.
*/
@OverrideImplementor(JVMMetricsSender.class)
-public class KafkaJVMMetricsSender extends JVMMetricsSender {
+public class KafkaJVMMetricsSender extends JVMMetricsSender implements
KafkaConnectionStatusListener {
private static final ILog LOGGER =
LogManager.getLogger(KafkaJVMMetricsSender.class);
private KafkaProducer<String, Bytes> producer;
private String topic;
-
private BlockingQueue<JVMMetric> queue;
- private volatile boolean running = false;
@Override
public void run() {
if (!queue.isEmpty()) {
List<JVMMetric> buffer = new ArrayList<>();
queue.drainTo(buffer);
- if (running) {
+ if (producer != null) {
JVMMetricCollection metrics = JVMMetricCollection.newBuilder()
.addAllMetrics(buffer)
.setService(Config.Agent.SERVICE_NAME)
.setServiceInstance(Config.Agent.INSTANCE_NAME)
.build();
- if (LOGGER.isDebugEnable()) {
- LOGGER.debug(
+ if (LOGGER.isDebugEnable()) {
+ LOGGER.debug(
"JVM metrics reporting, topic: {}, key: {}, length:
{}", topic, metrics.getServiceInstance(),
buffer.size()
- );
- }
+ );
+ }
- producer.send(new ProducerRecord<>(
+ producer.send(new ProducerRecord<>(
topic,
metrics.getServiceInstance(),
Bytes.wrap(metrics.toByteArray())
- ));
- producer.flush();
+ ));
+ producer.flush();
}
Review comment:
I will recheck later
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
