You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2021/08/05 08:02:03 UTC
[GitHub] [pulsar] Technoboy- opened a new pull request #11564: Add compacted topic metrics for TopicStats in CLI
Technoboy- opened a new pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564
### Motivation
Currently, there are no relative metrics about compacted topic, which would be extremely useful while tracking potential flows or examining the overall condition of compacted topics .
So , I have added below metrics:
- compactedTopicSize : the size after compacted
- compactedTopicMsgCount : the message count after compacted
- compactedTopicRemovedEvents : how many events had been removed after compacted
These 3 metrics will be display in topic stats CLI :
```
./pulsar-admin topics stats persistent://tenant/ns/topic
```
### Documentation
This patch will add metrics in cli , which would generate doc automatically.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- removed a comment on pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- removed a comment on pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#issuecomment-903056619
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r689111490
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorMXBeanImpl.java
##########
@@ -0,0 +1,71 @@
+/**
+ * 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.pulsar.compaction;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
+
+public class CompactorMXBeanImpl implements CompactorMXBean {
+
+ private final ConcurrentHashMap<String, LongAdder> topicRemovedEventOps = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, Long> lastCompactSucceedTimestampOps = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, Long> lastCompactFailedTimestampOps = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, Long> lastCompactDurationTimeOps = new ConcurrentHashMap<>();
Review comment:
How about iterating the broker-owned topics with a fix delay to remove these topics?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
michaeljmarshall commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r691796316
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorMXBeanImpl.java
##########
@@ -0,0 +1,124 @@
+/**
+ * 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.pulsar.compaction;
+
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
+import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+
+public class CompactorMXBeanImpl implements CompactorMXBean {
+
+ private final ConcurrentHashMap<String, CompactRecord> compactRecordOps = new ConcurrentHashMap<>();
+
+ private final Optional<BrokerService> brokerService;
+ private final ScheduledExecutorService scheduler;
+
+ public CompactorMXBeanImpl(Optional<BrokerService> brokerService, ScheduledExecutorService scheduler) {
+ this.brokerService = brokerService;
+ this.scheduler = scheduler;
+ this.scheduler.scheduleAtFixedRate(this::cleanup, 2, 2, TimeUnit.HOURS);
Review comment:
I see that the `BrokerService` has a method to reactively remove a topic from its cache when appropriate, which has the affect of removing the metrics for that topic on that broker.
In an environment where topics are unloaded frequently, 2 hours time is a long time to have incorrect metrics hosted by a broker. It might be possible to update the `removeTopicFromCache` method by adding a method call like the following:
```java
pulsar.getCompactor().getStats().removeTopic(topicName)
```
In this way, we could remove this scheduled task and then we wouldn't have incorrect metrics.
##########
File path: pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/CompactStats.java
##########
@@ -0,0 +1,37 @@
+/**
+ * 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.pulsar.common.policies.data;
+
+/**
+ * Statistics about compact.
+ */
+public interface CompactStats {
Review comment:
Nit: I've been wondering if it is worth changing the name of these stats to `CompactionStats`. The documentation talks about `compaction`, and so I somewhat expect the metrics to have `compaction` in the metric's name. I can see why you chose `CompactStats`, but I also could see `CompactionStats` as a valid alternative. Do you have a strong opinion either way?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] gaoran10 commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
gaoran10 commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r689612419
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorMXBeanImpl.java
##########
@@ -0,0 +1,125 @@
+/**
+ * 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.pulsar.compaction;
+
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
+import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+
+public class CompactorMXBeanImpl implements CompactorMXBean {
+
+ private final ConcurrentHashMap<String, CompactRecord> compactRecordOps = new ConcurrentHashMap<>();
+
+ private final Optional<BrokerService> brokerService;
+ private final ScheduledExecutorService scheduler;
+
+ public CompactorMXBeanImpl(Optional<BrokerService> brokerService, ScheduledExecutorService scheduler) {
+ this.brokerService = brokerService;
+ this.scheduler = scheduler;
+ this.scheduler.scheduleAtFixedRate(this::cleanup, 2, 2, TimeUnit.HOURS);
+ }
+
+ public void cleanup() {
+ if (compactRecordOps.isEmpty()) {
+ return;
+ }
+ Set<String> topics = new HashSet<>();
+ if (brokerService.isPresent()) {
+ brokerService.get().forEachTopic(t -> {
+ if (t instanceof PersistentTopic) {
+ topics.add(t.getName());
+ }
+ });
+ }
+ if (!topics.isEmpty()) {
+ compactRecordOps.forEach((k, v) -> {
+ if (!topics.contains(k)) {
+ compactRecordOps.remove(k);
+ }
+ });
+ }
+ }
+
+ public void addCompactRemovedEvent(String topic) {
+ compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord()).addCompactRemovedEvent();
+ }
+
+ public void addCompactStartOp(String topic) {
+ compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord()).reset();
+ }
+
+ public void addCompactEndOp(String topic, boolean succeed) {
+ CompactRecord compactRecord = compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord());
+ compactRecord.lastCompactDurationTimeInMills = TimeUnit.NANOSECONDS.toMillis(
+ System.nanoTime() - compactRecord.lastCompactStartTimeOp);
+ compactRecord.lastCompactRemovedEventCount = compactRecord.lastCompactRemovedEventCountOp.longValue();
+ if (succeed) {
+ compactRecord.lastCompactSucceedTimestamp = System.currentTimeMillis();
+ } else {
+ compactRecord.lastCompactFailedTimestamp = System.currentTimeMillis();
+ }
+ }
+
+ @Override
+ public long getLastCompactRemovedEventCount(String topic) {
+ return compactRecordOps.getOrDefault(topic, new CompactRecord()).lastCompactRemovedEventCount;
+ }
+
+ @Override
+ public long getLastCompactSucceedTimestamp(String topic) {
+ return compactRecordOps.getOrDefault(topic, new CompactRecord()).lastCompactSucceedTimestamp;
+ }
+
+ @Override
+ public long getLastCompactFailedTimestamp(String topic) {
+ return compactRecordOps.getOrDefault(topic, new CompactRecord()).lastCompactFailedTimestamp;
+ }
+
+ @Override
+ public long getLastCompactDurationTimeInMills(String topic) {
+ return compactRecordOps.getOrDefault(topic, new CompactRecord()).lastCompactDurationTimeInMills;
+ }
+
+ static class CompactRecord {
+
+ private long lastCompactRemovedEventCount = 0L;
+ private long lastCompactSucceedTimestamp = 0L;
+ private long lastCompactFailedTimestamp = 0L;
+ private long lastCompactDurationTimeInMills = 0L;
+
+ //
Review comment:
Could you clean up the comment.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#issuecomment-903064630
/pulsarbot run-failure-checks
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#issuecomment-903056619
/pulsarbot run-failure-checks
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
michaeljmarshall commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r687238143
##########
File path: pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/CompactStats.java
##########
@@ -0,0 +1,34 @@
+/**
+ * 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.pulsar.common.policies.data;
Review comment:
I hadn't noticed that this file was in a different module than its implementation. I see now that this is following the pattern. Makes sense to me. Thanks for the explanation.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
##########
@@ -1327,7 +1327,11 @@ public Compactor newCompactor() throws PulsarServerException {
}
public synchronized Compactor getCompactor() throws PulsarServerException {
- if (this.compactor == null) {
+ return getCompactor(true);
+ }
+
+ public synchronized Compactor getCompactor(boolean initialized) throws PulsarServerException {
+ if (this.compactor == null && initialized) {
Review comment:
Nit: I think it might be clearer to name the parameter `shouldInitialize`, but that is just my preference.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r689102490
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorMXBean.java
##########
@@ -0,0 +1,51 @@
+/**
+ * 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.pulsar.compaction;
+
+import org.apache.bookkeeper.common.annotation.InterfaceAudience;
+import org.apache.bookkeeper.common.annotation.InterfaceStability;
+
+/**
+ * JMX Bean interface for Compactor stats.
+ */
+@InterfaceAudience.LimitedPrivate
+@InterfaceStability.Stable
+public interface CompactorMXBean {
+
+ /**
+ * @return the removed event count of last compact
+ */
+ long getLastCompactRemovedEventCount(String topic);
+
+ /**
+ * @return the timestamp of last succeed compact
+ */
+ long getLastCompactSucceedTimestamp(String topic);
+
+ /**
+ * @return the timestamp of last failed compact
+ */
+ long getLastCompactFailedTimestamp(String topic);
+
+ /**
+ * @return the duration time of last compact
+ */
+ long getLastCompactDurationTime(String topic);
Review comment:
Yes, I think so. Using nanoTime to record the start, and convert to mills in addCompactEndOp.
BTW, I will change the method to the above name.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r689089771
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorMXBean.java
##########
@@ -0,0 +1,51 @@
+/**
+ * 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.pulsar.compaction;
+
+import org.apache.bookkeeper.common.annotation.InterfaceAudience;
+import org.apache.bookkeeper.common.annotation.InterfaceStability;
+
+/**
+ * JMX Bean interface for Compactor stats.
+ */
+@InterfaceAudience.LimitedPrivate
+@InterfaceStability.Stable
+public interface CompactorMXBean {
+
+ /**
+ * @return the removed event count of last compact
+ */
+ long getLastCompactRemovedEventCount(String topic);
+
+ /**
+ * @return the timestamp of last succeed compact
+ */
+ long getLastCompactSucceedTimestamp(String topic);
+
+ /**
+ * @return the timestamp of last failed compact
+ */
+ long getLastCompactFailedTimestamp(String topic);
+
+ /**
+ * @return the duration time of last compact
+ */
+ long getLastCompactDurationTime(String topic);
Review comment:
```suggestion
long getLastCompactDurationTimeInMills(String topic);
```
I think milliseconds is enough here?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorMXBeanImpl.java
##########
@@ -0,0 +1,71 @@
+/**
+ * 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.pulsar.compaction;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
+
+public class CompactorMXBeanImpl implements CompactorMXBean {
+
+ private final ConcurrentHashMap<String, LongAdder> topicRemovedEventOps = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, Long> lastCompactSucceedTimestampOps = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, Long> lastCompactFailedTimestampOps = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, Long> lastCompactDurationTimeOps = new ConcurrentHashMap<>();
+
+ public void addCompactRemovedEvent(String topic) {
+ topicRemovedEventOps.computeIfAbsent(topic, k -> new LongAdder()).increment();
+ }
+
+ public void addCompactStartOp(String topic) {
+ topicRemovedEventOps.remove(topic);
+ lastCompactDurationTimeOps.put(topic, System.nanoTime());
Review comment:
If the compact task is not complete, we will get a very large value here.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorMXBeanImpl.java
##########
@@ -0,0 +1,71 @@
+/**
+ * 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.pulsar.compaction;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
+
+public class CompactorMXBeanImpl implements CompactorMXBean {
+
+ private final ConcurrentHashMap<String, LongAdder> topicRemovedEventOps = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, Long> lastCompactSucceedTimestampOps = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, Long> lastCompactFailedTimestampOps = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, Long> lastCompactDurationTimeOps = new ConcurrentHashMap<>();
Review comment:
How do we clean up the cache if the topic does not own by this broker.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
michaeljmarshall commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r686153350
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -1809,9 +1811,25 @@ public TopicStatsImpl getStats(boolean getPreciseBacklog, boolean subscriptionBa
stats.lastOffloadLedgerId = ledger.getLastOffloadedLedgerId();
stats.lastOffloadSuccessTimeStamp = ledger.getLastOffloadedSuccessTimestamp();
stats.lastOffloadFailureTimeStamp = ledger.getLastOffloadedFailureTimestamp();
+ stats.compactedTopicSize = getCompactorMXBean().map(stat -> stat.getCompactedTopicSize(topic)).orElse(0D);
+ stats.compactedTopicMsgCount = getCompactorMXBean().map(stat ->
+ stat.getCompactedTopicMsgCount(topic)).orElse(0L);
+ stats.compactedTopicRemovedEventCount = getCompactorMXBean().map(stat ->
+ stat.getCompactedTopicRemovedEventCount(topic)).orElse(0L);
+ log.info("stats : {}", stats);
return stats;
}
+ private Optional<CompactorMXBean> getCompactorMXBean() {
+ Compactor compactor = null;
+ try {
+ compactor = brokerService.pulsar().getCompactor();
Review comment:
@Technoboy- why did you mark this as resolved? It looks like the `getCompactor` method is still being called, so my concerns are still present.
##########
File path: pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/CompactStats.java
##########
@@ -0,0 +1,34 @@
+/**
+ * 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.pulsar.common.policies.data;
Review comment:
Any reason this isn't in the `org.apache.pulsar.common.policies.data.stats` package?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] gaoran10 commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
gaoran10 commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r689607897
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorMXBeanImpl.java
##########
@@ -0,0 +1,125 @@
+/**
+ * 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.pulsar.compaction;
+
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
+import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+
+public class CompactorMXBeanImpl implements CompactorMXBean {
+
+ private final ConcurrentHashMap<String, CompactRecord> compactRecordOps = new ConcurrentHashMap<>();
+
+ private final Optional<BrokerService> brokerService;
+ private final ScheduledExecutorService scheduler;
+
+ public CompactorMXBeanImpl(Optional<BrokerService> brokerService, ScheduledExecutorService scheduler) {
+ this.brokerService = brokerService;
+ this.scheduler = scheduler;
+ this.scheduler.scheduleAtFixedRate(this::cleanup, 2, 2, TimeUnit.HOURS);
+ }
+
+ public void cleanup() {
+ if (compactRecordOps.isEmpty()) {
+ return;
+ }
+ Set<String> topics = new HashSet<>();
+ if (brokerService.isPresent()) {
+ brokerService.get().forEachTopic(t -> {
+ if (t instanceof PersistentTopic) {
+ topics.add(t.getName());
+ }
+ });
+ }
+ if (!topics.isEmpty()) {
+ compactRecordOps.forEach((k, v) -> {
+ if (!topics.contains(k)) {
+ compactRecordOps.remove(k);
+ }
+ });
+ }
+ }
+
+ public void addCompactRemovedEvent(String topic) {
+ compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord()).addCompactRemovedEvent();
+ }
+
+ public void addCompactStartOp(String topic) {
+ compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord()).reset();
+ }
+
+ public void addCompactEndOp(String topic, boolean succeed) {
+ CompactRecord compactRecord = compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord());
Review comment:
It seems that if the `CompactRecord` of the topic does not exist, it's an unusual situation, the stat is invalid, maybe we could log a warn log instead of recording the stat. One possible scenario is that the topic was deleted.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -1903,9 +1905,28 @@ public TopicStatsImpl getStats(boolean getPreciseBacklog, boolean subscriptionBa
stats.lastOffloadLedgerId = ledger.getLastOffloadedLedgerId();
stats.lastOffloadSuccessTimeStamp = ledger.getLastOffloadedSuccessTimestamp();
stats.lastOffloadFailureTimeStamp = ledger.getLastOffloadedFailureTimestamp();
+ stats.compact.lastCompactRemovedEventCount = getCompactorMXBean().map(stat ->
+ stat.getLastCompactRemovedEventCount(topic)).orElse(0L);
+ stats.compact.lastCompactSucceedTimestamp = getCompactorMXBean().map(stat ->
Review comment:
A trivial question, maybe we could reuse the `getCompactorMXBean()` result, call the method `getCompactorMXBean()` one time.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java
##########
@@ -145,14 +149,17 @@ private void phaseOneLoop(RawReader reader,
Pair<String, Integer> keyAndSize = extractKeyAndSize(m);
if (keyAndSize != null) {
if (keyAndSize.getRight() > 0) {
- latestForKey.put(keyAndSize.getLeft(), id);
+ MessageId old = latestForKey.put(keyAndSize.getLeft(), id);
+ replaceMessage = old != null;
} else {
deletedMessage = true;
latestForKey.remove(keyAndSize.getLeft());
}
}
}
-
+ if (replaceMessage || deletedMessage) {
+ mxBean.addCompactRemovedEvent(reader.getTopic());
+ }
Review comment:
Maybe the `replaceMessage` could be re-modified. For example, in a batch loop, at first, it could be true, when meeting a new key it will change to false.
I'm not sure about the meaning of the `compactRemoveEvent`, this event indicates one batch compact event or message compact events.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorMXBeanImpl.java
##########
@@ -0,0 +1,125 @@
+/**
+ * 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.pulsar.compaction;
+
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
+import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+
+public class CompactorMXBeanImpl implements CompactorMXBean {
+
+ private final ConcurrentHashMap<String, CompactRecord> compactRecordOps = new ConcurrentHashMap<>();
+
+ private final Optional<BrokerService> brokerService;
+ private final ScheduledExecutorService scheduler;
+
+ public CompactorMXBeanImpl(Optional<BrokerService> brokerService, ScheduledExecutorService scheduler) {
+ this.brokerService = brokerService;
+ this.scheduler = scheduler;
+ this.scheduler.scheduleAtFixedRate(this::cleanup, 2, 2, TimeUnit.HOURS);
+ }
+
+ public void cleanup() {
+ if (compactRecordOps.isEmpty()) {
+ return;
+ }
+ Set<String> topics = new HashSet<>();
+ if (brokerService.isPresent()) {
+ brokerService.get().forEachTopic(t -> {
+ if (t instanceof PersistentTopic) {
+ topics.add(t.getName());
+ }
+ });
+ }
+ if (!topics.isEmpty()) {
+ compactRecordOps.forEach((k, v) -> {
+ if (!topics.contains(k)) {
+ compactRecordOps.remove(k);
+ }
+ });
+ }
+ }
+
+ public void addCompactRemovedEvent(String topic) {
+ compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord()).addCompactRemovedEvent();
+ }
+
+ public void addCompactStartOp(String topic) {
+ compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord()).reset();
+ }
+
+ public void addCompactEndOp(String topic, boolean succeed) {
+ CompactRecord compactRecord = compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord());
+ compactRecord.lastCompactDurationTimeInMills = TimeUnit.NANOSECONDS.toMillis(
+ System.nanoTime() - compactRecord.lastCompactStartTimeOp);
+ compactRecord.lastCompactRemovedEventCount = compactRecord.lastCompactRemovedEventCountOp.longValue();
+ if (succeed) {
+ compactRecord.lastCompactSucceedTimestamp = System.currentTimeMillis();
+ } else {
+ compactRecord.lastCompactFailedTimestamp = System.currentTimeMillis();
+ }
+ }
+
+ @Override
+ public long getLastCompactRemovedEventCount(String topic) {
+ return compactRecordOps.getOrDefault(topic, new CompactRecord()).lastCompactRemovedEventCount;
+ }
+
+ @Override
+ public long getLastCompactSucceedTimestamp(String topic) {
+ return compactRecordOps.getOrDefault(topic, new CompactRecord()).lastCompactSucceedTimestamp;
+ }
+
+ @Override
+ public long getLastCompactFailedTimestamp(String topic) {
+ return compactRecordOps.getOrDefault(topic, new CompactRecord()).lastCompactFailedTimestamp;
+ }
+
+ @Override
+ public long getLastCompactDurationTimeInMills(String topic) {
+ return compactRecordOps.getOrDefault(topic, new CompactRecord()).lastCompactDurationTimeInMills;
+ }
+
+ static class CompactRecord {
+
+ private long lastCompactRemovedEventCount = 0L;
+ private long lastCompactSucceedTimestamp = 0L;
+ private long lastCompactFailedTimestamp = 0L;
+ private long lastCompactDurationTimeInMills = 0L;
+
+ //
Review comment:
Could you clean up the comment?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r686422506
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -1809,9 +1811,25 @@ public TopicStatsImpl getStats(boolean getPreciseBacklog, boolean subscriptionBa
stats.lastOffloadLedgerId = ledger.getLastOffloadedLedgerId();
stats.lastOffloadSuccessTimeStamp = ledger.getLastOffloadedSuccessTimestamp();
stats.lastOffloadFailureTimeStamp = ledger.getLastOffloadedFailureTimestamp();
+ stats.compactedTopicSize = getCompactorMXBean().map(stat -> stat.getCompactedTopicSize(topic)).orElse(0D);
+ stats.compactedTopicMsgCount = getCompactorMXBean().map(stat ->
+ stat.getCompactedTopicMsgCount(topic)).orElse(0L);
+ stats.compactedTopicRemovedEventCount = getCompactorMXBean().map(stat ->
+ stat.getCompactedTopicRemovedEventCount(topic)).orElse(0L);
+ log.info("stats : {}", stats);
return stats;
}
+ private Optional<CompactorMXBean> getCompactorMXBean() {
+ Compactor compactor = null;
+ try {
+ compactor = brokerService.pulsar().getCompactor();
Review comment:
Sorry, mistake for this, I will focus on this.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#issuecomment-901946685
@michaeljmarshall @gaoran10 @codelipenghui thanks for reviewing.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#issuecomment-896023532
/pulsarbot run-failure-checks
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r686423836
##########
File path: pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/CompactStats.java
##########
@@ -0,0 +1,34 @@
+/**
+ * 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.pulsar.common.policies.data;
Review comment:
I put it in the `data.stats` at first, but found that all other interfaces like PublisherStats, ConsumerStats were put in `org.apache.pulsar.common.policies`. We may discuss this.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
michaeljmarshall commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r685363477
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorMXBeanImpl.java
##########
@@ -0,0 +1,78 @@
+/**
+ * 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.pulsar.compaction;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.LongAdder;
+import org.apache.pulsar.common.stats.Sum;
+
+public class CompactorMXBeanImpl implements CompactorMXBean {
+
+ private final ConcurrentMap<String, Sum> sumOps = new ConcurrentHashMap<>();
+
+ private final ConcurrentHashMap<String, Integer> actionOps = new ConcurrentHashMap();
Review comment:
What is the purpose of this map? I see key value pairs getting updated, but never retrieved.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorMXBeanImpl.java
##########
@@ -0,0 +1,78 @@
+/**
+ * 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.pulsar.compaction;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.LongAdder;
+import org.apache.pulsar.common.stats.Sum;
+
+public class CompactorMXBeanImpl implements CompactorMXBean {
+
+ private final ConcurrentMap<String, Sum> sumOps = new ConcurrentHashMap<>();
+
+ private final ConcurrentHashMap<String, Integer> actionOps = new ConcurrentHashMap();
+ private final ConcurrentHashMap<String, LongAdder> removedOps = new ConcurrentHashMap<>();
+
+ public void addCompactSample(String topic, int size) {
+ sumOps.compute(topic, (k, v) -> {
+ if (v == null) {
+ v = new Sum();
+ }
+ v.recordEvent(size);
+ return v;
+ });
Review comment:
```suggestion
sumOps.computeIfAbsent(topic, k -> new Sum()).recordEvent(size);
```
##########
File path: pulsar-common/src/main/java/org/apache/pulsar/common/stats/Sum.java
##########
@@ -0,0 +1,47 @@
+/**
+ * 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.pulsar.common.stats;
+
+import java.util.concurrent.atomic.LongAdder;
+
+/**
+ */
Review comment:
It'd be nice to add documentation here. For example, this class is intended to be thread safe, and a comment stating so would be good. Additionally, documenting when to use this class for metrics would be helpful.
##########
File path: pulsar-common/src/main/java/org/apache/pulsar/common/stats/Sum.java
##########
@@ -0,0 +1,47 @@
+/**
+ * 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.pulsar.common.stats;
+
+import java.util.concurrent.atomic.LongAdder;
+
+/**
+ */
+public class Sum {
Review comment:
This `Sum` class should have associated tests.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorMXBeanImpl.java
##########
@@ -0,0 +1,78 @@
+/**
+ * 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.pulsar.compaction;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.LongAdder;
+import org.apache.pulsar.common.stats.Sum;
+
+public class CompactorMXBeanImpl implements CompactorMXBean {
+
+ private final ConcurrentMap<String, Sum> sumOps = new ConcurrentHashMap<>();
+
+ private final ConcurrentHashMap<String, Integer> actionOps = new ConcurrentHashMap();
+ private final ConcurrentHashMap<String, LongAdder> removedOps = new ConcurrentHashMap<>();
+
+ public void addCompactSample(String topic, int size) {
+ sumOps.compute(topic, (k, v) -> {
+ if (v == null) {
+ v = new Sum();
+ }
+ v.recordEvent(size);
+ return v;
+ });
+ }
+
+ public void addCompactRemovedEvent(String topic) {
+ removedOps.compute(topic, (k, v) -> {
+ if (v == null) {
+ v = new LongAdder();
+ }
+ v.increment();
+ return v;
+ });
Review comment:
```suggestion
removedOps.computeIfAbsent(topic, k -> new LongAdder()).increment();
```
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r689972100
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorMXBeanImpl.java
##########
@@ -0,0 +1,125 @@
+/**
+ * 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.pulsar.compaction;
+
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
+import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+
+public class CompactorMXBeanImpl implements CompactorMXBean {
+
+ private final ConcurrentHashMap<String, CompactRecord> compactRecordOps = new ConcurrentHashMap<>();
+
+ private final Optional<BrokerService> brokerService;
+ private final ScheduledExecutorService scheduler;
+
+ public CompactorMXBeanImpl(Optional<BrokerService> brokerService, ScheduledExecutorService scheduler) {
+ this.brokerService = brokerService;
+ this.scheduler = scheduler;
+ this.scheduler.scheduleAtFixedRate(this::cleanup, 2, 2, TimeUnit.HOURS);
+ }
+
+ public void cleanup() {
+ if (compactRecordOps.isEmpty()) {
+ return;
+ }
+ Set<String> topics = new HashSet<>();
+ if (brokerService.isPresent()) {
+ brokerService.get().forEachTopic(t -> {
+ if (t instanceof PersistentTopic) {
+ topics.add(t.getName());
+ }
+ });
+ }
+ if (!topics.isEmpty()) {
+ compactRecordOps.forEach((k, v) -> {
+ if (!topics.contains(k)) {
+ compactRecordOps.remove(k);
+ }
+ });
+ }
+ }
+
+ public void addCompactRemovedEvent(String topic) {
+ compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord()).addCompactRemovedEvent();
+ }
+
+ public void addCompactStartOp(String topic) {
+ compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord()).reset();
+ }
+
+ public void addCompactEndOp(String topic, boolean succeed) {
+ CompactRecord compactRecord = compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord());
Review comment:
Yes, we treat this scenario by removing the topics. We will do the removing operation in `cleanup` periodically.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
michaeljmarshall commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r685346289
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -1809,9 +1811,25 @@ public TopicStatsImpl getStats(boolean getPreciseBacklog, boolean subscriptionBa
stats.lastOffloadLedgerId = ledger.getLastOffloadedLedgerId();
stats.lastOffloadSuccessTimeStamp = ledger.getLastOffloadedSuccessTimestamp();
stats.lastOffloadFailureTimeStamp = ledger.getLastOffloadedFailureTimestamp();
+ stats.compactedTopicSize = getCompactorMXBean().map(stat -> stat.getCompactedTopicSize(topic)).orElse(0D);
+ stats.compactedTopicMsgCount = getCompactorMXBean().map(stat ->
+ stat.getCompactedTopicMsgCount(topic)).orElse(0L);
+ stats.compactedTopicRemovedEventCount = getCompactorMXBean().map(stat ->
+ stat.getCompactedTopicRemovedEventCount(topic)).orElse(0L);
+ log.info("stats : {}", stats);
return stats;
}
+ private Optional<CompactorMXBean> getCompactorMXBean() {
+ Compactor compactor = null;
+ try {
+ compactor = brokerService.pulsar().getCompactor();
Review comment:
It looks like the call to `getCompactor` initializes a `TwoPhaseCompactor` class, if it hasn't already been initialized. Currently, the class is only initialized for a broker when a topic hosted by the broker has compaction turned on. Wouldn't we want to avoid initializing this class, if possible? One way to do this would be to add a method to the `PulsarService` class that optionally gets the compactor or returns an empty option if it hasn't been initialized.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -1809,9 +1811,25 @@ public TopicStatsImpl getStats(boolean getPreciseBacklog, boolean subscriptionBa
stats.lastOffloadLedgerId = ledger.getLastOffloadedLedgerId();
stats.lastOffloadSuccessTimeStamp = ledger.getLastOffloadedSuccessTimestamp();
stats.lastOffloadFailureTimeStamp = ledger.getLastOffloadedFailureTimestamp();
+ stats.compactedTopicSize = getCompactorMXBean().map(stat -> stat.getCompactedTopicSize(topic)).orElse(0D);
+ stats.compactedTopicMsgCount = getCompactorMXBean().map(stat ->
+ stat.getCompactedTopicMsgCount(topic)).orElse(0L);
+ stats.compactedTopicRemovedEventCount = getCompactorMXBean().map(stat ->
+ stat.getCompactedTopicRemovedEventCount(topic)).orElse(0L);
+ log.info("stats : {}", stats);
Review comment:
This log line seems like it could be removed.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui merged pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
codelipenghui merged pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r691963717
##########
File path: pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/CompactStats.java
##########
@@ -0,0 +1,37 @@
+/**
+ * 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.pulsar.common.policies.data;
+
+/**
+ * Statistics about compact.
+ */
+public interface CompactStats {
Review comment:
Yes, I will change to CompactionStats.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#issuecomment-903052929
```
Error: Tests run: 28, Failures: 12, Errors: 0, Skipped: 13, Time elapsed: 601.287 s <<< FAILURE! - in org.apache.pulsar.functions.source.batch.BatchSourceExecutorTest
Error: testPushLifeCycle(org.apache.pulsar.functions.source.batch.BatchSourceExecutorTest) Time elapsed: 300.011 s <<< FAILURE!
org.testng.internal.thread.ThreadTimeoutException: Method org.apache.pulsar.functions.source.batch.BatchSourceExecutorTest.testPushLifeCycle() didn't finish within the time-out 300000
at org.testng.internal.MethodInvocationHelper.invokeWithTimeoutWithNewExecutor(MethodInvocationHelper.java:371)
at org.testng.internal.MethodInvocationHelper.invokeWithTimeout(MethodInvocationHelper.java:282)
at org.testng.internal.TestInvoker.invokeMethod(TestInvoker.java:605)
at org.testng.internal.TestInvoker.retryFailed(TestInvoker.java:214)
at org.testng.internal.MethodRunner.runInSequence(MethodRunner.java:58)
at org.testng.internal.TestInvoker$MethodInvocationAgent.invoke(TestInvoker.java:822)
at org.testng.internal.TestInvoker.invokeTestMethods(TestInvoker.java:147)
at org.testng.internal.TestMethodWorker.invokeTestMethods(TestMethodWorker.java:146)
at org.testng.internal.TestMethodWorker.run(TestMethodWorker.java:128)
at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)
at org.testng.TestRunner.privateRun(TestRunner.java:764)
at org.testng.TestRunner.run(TestRunner.java:585)
at org.testng.SuiteRunner.runTest(SuiteRunner.java:384)
at org.testng.SuiteRunner.runSequentially(SuiteRunner.java:378)
at org.testng.SuiteRunner.privateRun(SuiteRunner.java:337)
at org.testng.SuiteRunner.run(SuiteRunner.java:286)
at org.testng.SuiteRunnerWorker.runSuite(SuiteRunnerWorker.java:53)
at org.testng.SuiteRunnerWorker.run(SuiteRunnerWorker.java:96)
at org.testng.TestNG.runSuitesSequentially(TestNG.java:1218)
at org.testng.TestNG.runSuitesLocally(TestNG.java:1140)
at org.testng.TestNG.runSuites(TestNG.java:1069)
at org.testng.TestNG.run(TestNG.java:1037)
at org.apache.maven.surefire.testng.TestNGExecutor.run(TestNGExecutor.java:135)
at org.apache.maven.surefire.testng.TestNGDirectoryTestSuite.executeSingleClass(TestNGDirectoryTestSuite.java:112)
at org.apache.maven.surefire.testng.TestNGDirectoryTestSuite.executeLazy(TestNGDirectoryTestSuite.java:123)
at org.apache.maven.surefire.testng.TestNGDirectoryTestSuite.execute(TestNGDirectoryTestSuite.java:90)
at org.apache.maven.surefire.testng.TestNGProvider.invoke(TestNGProvider.java:146)
at org.apache.maven.surefire.booter.ForkedBooter.invokeProviderInSameClassLoader(ForkedBooter.java:384)
at org.apache.maven.surefire.booter.ForkedBooter.runSuitesInProcess(ForkedBooter.java:345)
at org.apache.maven.surefire.booter.ForkedBooter.execute(ForkedBooter.java:126)
at org.apache.maven.surefire.booter.ForkedBooter.main(ForkedBooter.java:418)
```
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
michaeljmarshall commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r686152453
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -1809,9 +1811,25 @@ public TopicStatsImpl getStats(boolean getPreciseBacklog, boolean subscriptionBa
stats.lastOffloadLedgerId = ledger.getLastOffloadedLedgerId();
stats.lastOffloadSuccessTimeStamp = ledger.getLastOffloadedSuccessTimestamp();
stats.lastOffloadFailureTimeStamp = ledger.getLastOffloadedFailureTimestamp();
+ stats.compactedTopicSize = getCompactorMXBean().map(stat -> stat.getCompactedTopicSize(topic)).orElse(0D);
+ stats.compactedTopicMsgCount = getCompactorMXBean().map(stat ->
+ stat.getCompactedTopicMsgCount(topic)).orElse(0L);
+ stats.compactedTopicRemovedEventCount = getCompactorMXBean().map(stat ->
+ stat.getCompactedTopicRemovedEventCount(topic)).orElse(0L);
+ log.info("stats : {}", stats);
return stats;
}
+ private Optional<CompactorMXBean> getCompactorMXBean() {
+ Compactor compactor = null;
+ try {
+ compactor = brokerService.pulsar().getCompactor();
Review comment:
@Technoboy- why did this get resolved? It looks like the `getCompactor` method is still being called, so my concerns are still present.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#issuecomment-900768221
@michaeljmarshall Please help review the PR again.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r693290971
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorMXBeanImpl.java
##########
@@ -0,0 +1,93 @@
+/**
+ * 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.pulsar.compaction;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
+
+public class CompactorMXBeanImpl implements CompactorMXBean {
+
+ private final ConcurrentHashMap<String, CompactRecord> compactRecordOps = new ConcurrentHashMap<>();
+
+ public void addCompactRemovedEvent(String topic) {
+ compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord()).addCompactRemovedEvent();
+ }
+
+ public void addCompactStartOp(String topic) {
+ compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord()).reset();
+ }
+
+ public void addCompactEndOp(String topic, boolean succeed) {
+ CompactRecord compactRecord = compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord());
+ compactRecord.lastCompactDurationTimeInMills = TimeUnit.NANOSECONDS.toMillis(
+ System.nanoTime() - compactRecord.lastCompactStartTimeOp);
Review comment:
Ok, thanks for kindly review.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorMXBeanImpl.java
##########
@@ -0,0 +1,93 @@
+/**
+ * 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.pulsar.compaction;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
+
+public class CompactorMXBeanImpl implements CompactorMXBean {
+
+ private final ConcurrentHashMap<String, CompactRecord> compactRecordOps = new ConcurrentHashMap<>();
+
+ public void addCompactRemovedEvent(String topic) {
+ compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord()).addCompactRemovedEvent();
+ }
+
+ public void addCompactStartOp(String topic) {
+ compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord()).reset();
+ }
+
+ public void addCompactEndOp(String topic, boolean succeed) {
+ CompactRecord compactRecord = compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord());
+ compactRecord.lastCompactDurationTimeInMills = TimeUnit.NANOSECONDS.toMillis(
+ System.nanoTime() - compactRecord.lastCompactStartTimeOp);
Review comment:
Ok, thanks for the kindly review.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r693290940
##########
File path: pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/CompactionStats.java
##########
@@ -0,0 +1,37 @@
+/**
+ * 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.pulsar.common.policies.data;
+
+/**
+ * Statistics about compaction.
+ */
+public interface CompactionStats {
+
+ /** The removed event count of last compact. */
+ long getLastCompactRemovedEventCount();
+
+ /** The timestamp of last succeed compact. */
+ long getLastCompactSucceedTimestamp();
+
+ /** The timestamp of last failed compact. */
+ long getLastCompactFailedTimestamp();
+
+ /** The duration time of last compact. */
+ long getLastCompactDurationTimeInMills();
Review comment:
Yes, thanks for the kindly review. I will update now.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r688603056
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorMXBeanImpl.java
##########
@@ -0,0 +1,60 @@
+/**
+ * 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.pulsar.compaction;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.LongAdder;
+
+public class CompactorMXBeanImpl implements CompactorMXBean {
+
+ private final ConcurrentHashMap<String, LongAdder> topicRemovedEventOps = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, Long> lastCompactSucceedTimestampOps = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, Long> lastCompactFailedTimestampOps = new ConcurrentHashMap<>();
+
+ public void addCompactRemovedEvent(String topic) {
+ topicRemovedEventOps.computeIfAbsent(topic, k -> new LongAdder()).increment();
+ }
+
+ public void addCompactStartOp(String topic) {
Review comment:
Ok, Add lastCompactDurationTime.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#issuecomment-894155101
/pulsarbot run-failure-checks
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#issuecomment-901612365
@michaeljmarshall Please help review the PR again.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r686503868
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -1809,9 +1811,25 @@ public TopicStatsImpl getStats(boolean getPreciseBacklog, boolean subscriptionBa
stats.lastOffloadLedgerId = ledger.getLastOffloadedLedgerId();
stats.lastOffloadSuccessTimeStamp = ledger.getLastOffloadedSuccessTimestamp();
stats.lastOffloadFailureTimeStamp = ledger.getLastOffloadedFailureTimestamp();
+ stats.compactedTopicSize = getCompactorMXBean().map(stat -> stat.getCompactedTopicSize(topic)).orElse(0D);
+ stats.compactedTopicMsgCount = getCompactorMXBean().map(stat ->
+ stat.getCompactedTopicMsgCount(topic)).orElse(0L);
+ stats.compactedTopicRemovedEventCount = getCompactorMXBean().map(stat ->
+ stat.getCompactedTopicRemovedEventCount(topic)).orElse(0L);
+ log.info("stats : {}", stats);
return stats;
}
+ private Optional<CompactorMXBean> getCompactorMXBean() {
+ Compactor compactor = null;
+ try {
+ compactor = brokerService.pulsar().getCompactor();
Review comment:
Hi @michaeljmarshall , I have added a new method to avoid initializing the compactor. Could you help to review ?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r688176169
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorMXBeanImpl.java
##########
@@ -0,0 +1,60 @@
+/**
+ * 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.pulsar.compaction;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.LongAdder;
+
+public class CompactorMXBeanImpl implements CompactorMXBean {
+
+ private final ConcurrentHashMap<String, LongAdder> topicRemovedEventOps = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, Long> lastCompactSucceedTimestampOps = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, Long> lastCompactFailedTimestampOps = new ConcurrentHashMap<>();
+
+ public void addCompactRemovedEvent(String topic) {
+ topicRemovedEventOps.computeIfAbsent(topic, k -> new LongAdder()).increment();
+ }
+
+ public void addCompactStartOp(String topic) {
Review comment:
We have the start timestamp and the end timestamp, it's great to have a time duration for the last compaction task.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r689111684
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorMXBeanImpl.java
##########
@@ -0,0 +1,71 @@
+/**
+ * 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.pulsar.compaction;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
+
+public class CompactorMXBeanImpl implements CompactorMXBean {
+
+ private final ConcurrentHashMap<String, LongAdder> topicRemovedEventOps = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, Long> lastCompactSucceedTimestampOps = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, Long> lastCompactFailedTimestampOps = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, Long> lastCompactDurationTimeOps = new ConcurrentHashMap<>();
+
+ public void addCompactRemovedEvent(String topic) {
+ topicRemovedEventOps.computeIfAbsent(topic, k -> new LongAdder()).increment();
+ }
+
+ public void addCompactStartOp(String topic) {
+ topicRemovedEventOps.remove(topic);
+ lastCompactDurationTimeOps.put(topic, System.nanoTime());
Review comment:
I have fixed this issue by adding a new field to mark the start time in a later update.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r689964478
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java
##########
@@ -145,14 +149,17 @@ private void phaseOneLoop(RawReader reader,
Pair<String, Integer> keyAndSize = extractKeyAndSize(m);
if (keyAndSize != null) {
if (keyAndSize.getRight() > 0) {
- latestForKey.put(keyAndSize.getLeft(), id);
+ MessageId old = latestForKey.put(keyAndSize.getLeft(), id);
+ replaceMessage = old != null;
} else {
deletedMessage = true;
latestForKey.remove(keyAndSize.getLeft());
}
}
}
-
+ if (replaceMessage || deletedMessage) {
+ mxBean.addCompactRemovedEvent(reader.getTopic());
+ }
Review comment:
`compactRemoveEvent` indicates how many duplicate messages have been removed in this compact task.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r689963705
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorMXBeanImpl.java
##########
@@ -0,0 +1,125 @@
+/**
+ * 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.pulsar.compaction;
+
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
+import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+
+public class CompactorMXBeanImpl implements CompactorMXBean {
+
+ private final ConcurrentHashMap<String, CompactRecord> compactRecordOps = new ConcurrentHashMap<>();
+
+ private final Optional<BrokerService> brokerService;
+ private final ScheduledExecutorService scheduler;
+
+ public CompactorMXBeanImpl(Optional<BrokerService> brokerService, ScheduledExecutorService scheduler) {
+ this.brokerService = brokerService;
+ this.scheduler = scheduler;
+ this.scheduler.scheduleAtFixedRate(this::cleanup, 2, 2, TimeUnit.HOURS);
+ }
+
+ public void cleanup() {
+ if (compactRecordOps.isEmpty()) {
+ return;
+ }
+ Set<String> topics = new HashSet<>();
+ if (brokerService.isPresent()) {
+ brokerService.get().forEachTopic(t -> {
+ if (t instanceof PersistentTopic) {
+ topics.add(t.getName());
+ }
+ });
+ }
+ if (!topics.isEmpty()) {
+ compactRecordOps.forEach((k, v) -> {
+ if (!topics.contains(k)) {
+ compactRecordOps.remove(k);
+ }
+ });
+ }
+ }
+
+ public void addCompactRemovedEvent(String topic) {
+ compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord()).addCompactRemovedEvent();
+ }
+
+ public void addCompactStartOp(String topic) {
+ compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord()).reset();
+ }
+
+ public void addCompactEndOp(String topic, boolean succeed) {
+ CompactRecord compactRecord = compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord());
+ compactRecord.lastCompactDurationTimeInMills = TimeUnit.NANOSECONDS.toMillis(
+ System.nanoTime() - compactRecord.lastCompactStartTimeOp);
+ compactRecord.lastCompactRemovedEventCount = compactRecord.lastCompactRemovedEventCountOp.longValue();
+ if (succeed) {
+ compactRecord.lastCompactSucceedTimestamp = System.currentTimeMillis();
+ } else {
+ compactRecord.lastCompactFailedTimestamp = System.currentTimeMillis();
+ }
+ }
+
+ @Override
+ public long getLastCompactRemovedEventCount(String topic) {
+ return compactRecordOps.getOrDefault(topic, new CompactRecord()).lastCompactRemovedEventCount;
+ }
+
+ @Override
+ public long getLastCompactSucceedTimestamp(String topic) {
+ return compactRecordOps.getOrDefault(topic, new CompactRecord()).lastCompactSucceedTimestamp;
+ }
+
+ @Override
+ public long getLastCompactFailedTimestamp(String topic) {
+ return compactRecordOps.getOrDefault(topic, new CompactRecord()).lastCompactFailedTimestamp;
+ }
+
+ @Override
+ public long getLastCompactDurationTimeInMills(String topic) {
+ return compactRecordOps.getOrDefault(topic, new CompactRecord()).lastCompactDurationTimeInMills;
+ }
+
+ static class CompactRecord {
+
+ private long lastCompactRemovedEventCount = 0L;
+ private long lastCompactSucceedTimestamp = 0L;
+ private long lastCompactFailedTimestamp = 0L;
+ private long lastCompactDurationTimeInMills = 0L;
+
+ //
Review comment:
ok.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#issuecomment-903119721
/pulsarbot run-failure-checks
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#issuecomment-896554264
/pulsarbot run-failure-checks
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on pull request #11564: [WIP]Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#issuecomment-895654595
> I suggested a few changes and asked some questions about design. I think the `Sum` class needs tests before this can be merged.
Thanks for reviewing. Yesterday, we find that compacted topic has metrics in InternalStats which would be shown by:
```
./pulsar-admin topics stats-internal persistent://tenant/ns/topic
```
I may have to record other metrics except the above two `compactedTopicSize `,`compactedTopicMsgCount `
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r689967367
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -1903,9 +1905,28 @@ public TopicStatsImpl getStats(boolean getPreciseBacklog, boolean subscriptionBa
stats.lastOffloadLedgerId = ledger.getLastOffloadedLedgerId();
stats.lastOffloadSuccessTimeStamp = ledger.getLastOffloadedSuccessTimestamp();
stats.lastOffloadFailureTimeStamp = ledger.getLastOffloadedFailureTimestamp();
+ stats.compact.lastCompactRemovedEventCount = getCompactorMXBean().map(stat ->
+ stat.getLastCompactRemovedEventCount(topic)).orElse(0L);
+ stats.compact.lastCompactSucceedTimestamp = getCompactorMXBean().map(stat ->
Review comment:
Ok.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r691985618
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorMXBeanImpl.java
##########
@@ -0,0 +1,124 @@
+/**
+ * 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.pulsar.compaction;
+
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
+import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+
+public class CompactorMXBeanImpl implements CompactorMXBean {
+
+ private final ConcurrentHashMap<String, CompactRecord> compactRecordOps = new ConcurrentHashMap<>();
+
+ private final Optional<BrokerService> brokerService;
+ private final ScheduledExecutorService scheduler;
+
+ public CompactorMXBeanImpl(Optional<BrokerService> brokerService, ScheduledExecutorService scheduler) {
+ this.brokerService = brokerService;
+ this.scheduler = scheduler;
+ this.scheduler.scheduleAtFixedRate(this::cleanup, 2, 2, TimeUnit.HOURS);
Review comment:
Thanks for reviewing and update according to your suggestion.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r689972100
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorMXBeanImpl.java
##########
@@ -0,0 +1,125 @@
+/**
+ * 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.pulsar.compaction;
+
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
+import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+
+public class CompactorMXBeanImpl implements CompactorMXBean {
+
+ private final ConcurrentHashMap<String, CompactRecord> compactRecordOps = new ConcurrentHashMap<>();
+
+ private final Optional<BrokerService> brokerService;
+ private final ScheduledExecutorService scheduler;
+
+ public CompactorMXBeanImpl(Optional<BrokerService> brokerService, ScheduledExecutorService scheduler) {
+ this.brokerService = brokerService;
+ this.scheduler = scheduler;
+ this.scheduler.scheduleAtFixedRate(this::cleanup, 2, 2, TimeUnit.HOURS);
+ }
+
+ public void cleanup() {
+ if (compactRecordOps.isEmpty()) {
+ return;
+ }
+ Set<String> topics = new HashSet<>();
+ if (brokerService.isPresent()) {
+ brokerService.get().forEachTopic(t -> {
+ if (t instanceof PersistentTopic) {
+ topics.add(t.getName());
+ }
+ });
+ }
+ if (!topics.isEmpty()) {
+ compactRecordOps.forEach((k, v) -> {
+ if (!topics.contains(k)) {
+ compactRecordOps.remove(k);
+ }
+ });
+ }
+ }
+
+ public void addCompactRemovedEvent(String topic) {
+ compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord()).addCompactRemovedEvent();
+ }
+
+ public void addCompactStartOp(String topic) {
+ compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord()).reset();
+ }
+
+ public void addCompactEndOp(String topic, boolean succeed) {
+ CompactRecord compactRecord = compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord());
Review comment:
Yes, we treat this scenario by removing the topics. We will do this in `cleanup` periodically.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#issuecomment-900768221
@michaeljmarshall Please help review the PR again.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#issuecomment-903113114
/pulsarbot run-failure-checks
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r687285107
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
##########
@@ -1327,7 +1327,11 @@ public Compactor newCompactor() throws PulsarServerException {
}
public synchronized Compactor getCompactor() throws PulsarServerException {
- if (this.compactor == null) {
+ return getCompactor(true);
+ }
+
+ public synchronized Compactor getCompactor(boolean initialized) throws PulsarServerException {
+ if (this.compactor == null && initialized) {
Review comment:
Ok, that's good. I have updated to this. Many thanks for reviewing.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
michaeljmarshall commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r693236571
##########
File path: pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/CompactionStats.java
##########
@@ -0,0 +1,37 @@
+/**
+ * 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.pulsar.common.policies.data;
+
+/**
+ * Statistics about compaction.
+ */
+public interface CompactionStats {
+
+ /** The removed event count of last compact. */
+ long getLastCompactRemovedEventCount();
+
+ /** The timestamp of last succeed compact. */
+ long getLastCompactSucceedTimestamp();
+
+ /** The timestamp of last failed compact. */
+ long getLastCompactFailedTimestamp();
+
+ /** The duration time of last compact. */
+ long getLastCompactDurationTimeInMills();
Review comment:
Since you updated `CompactStats` to `CompactionStats`, do you think we should also update these metrics from `Compact` to `Compaction`? I think I'd prefer `Compaction`, but I don't feel strongly. I just want to mention it before this gets merged.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorMXBeanImpl.java
##########
@@ -0,0 +1,93 @@
+/**
+ * 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.pulsar.compaction;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
+
+public class CompactorMXBeanImpl implements CompactorMXBean {
+
+ private final ConcurrentHashMap<String, CompactRecord> compactRecordOps = new ConcurrentHashMap<>();
+
+ public void addCompactRemovedEvent(String topic) {
+ compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord()).addCompactRemovedEvent();
+ }
+
+ public void addCompactStartOp(String topic) {
+ compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord()).reset();
+ }
+
+ public void addCompactEndOp(String topic, boolean succeed) {
+ CompactRecord compactRecord = compactRecordOps.computeIfAbsent(topic, k -> new CompactRecord());
+ compactRecord.lastCompactDurationTimeInMills = TimeUnit.NANOSECONDS.toMillis(
+ System.nanoTime() - compactRecord.lastCompactStartTimeOp);
Review comment:
Nit: why compute the difference in nano time if we're only going to capture millis in the metric? Seems like we could use `System.currentTimeMillis()` instead, and then we wouldn't need to convert time units here.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#issuecomment-902344809
/pulsarbot run-failure-checks
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r689967293
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java
##########
@@ -145,14 +149,17 @@ private void phaseOneLoop(RawReader reader,
Pair<String, Integer> keyAndSize = extractKeyAndSize(m);
if (keyAndSize != null) {
if (keyAndSize.getRight() > 0) {
- latestForKey.put(keyAndSize.getLeft(), id);
+ MessageId old = latestForKey.put(keyAndSize.getLeft(), id);
+ replaceMessage = old != null;
} else {
deletedMessage = true;
latestForKey.remove(keyAndSize.getLeft());
}
}
}
-
+ if (replaceMessage || deletedMessage) {
+ mxBean.addCompactRemovedEvent(reader.getTopic());
+ }
Review comment:
> Maybe the `replaceMessage` could be re-modified. For example, in a batch loop, at first, it could be true, when meeting a new key it will change to false.
Yes, really, Thanks for the kindly review.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #11564: Add compacted topic metrics for TopicStats in CLI
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #11564:
URL: https://github.com/apache/pulsar/pull/11564#discussion_r691825650
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorMXBeanImpl.java
##########
@@ -0,0 +1,124 @@
+/**
+ * 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.pulsar.compaction;
+
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
+import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+
+public class CompactorMXBeanImpl implements CompactorMXBean {
+
+ private final ConcurrentHashMap<String, CompactRecord> compactRecordOps = new ConcurrentHashMap<>();
+
+ private final Optional<BrokerService> brokerService;
+ private final ScheduledExecutorService scheduler;
+
+ public CompactorMXBeanImpl(Optional<BrokerService> brokerService, ScheduledExecutorService scheduler) {
+ this.brokerService = brokerService;
+ this.scheduler = scheduler;
+ this.scheduler.scheduleAtFixedRate(this::cleanup, 2, 2, TimeUnit.HOURS);
Review comment:
Yes, good idea. I think we can remove from the compaction stats map when closing the topic. @Technoboy-
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org