You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "poorbarcode (via GitHub)" <gi...@apache.org> on 2025/07/29 18:06:39 UTC
[PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
poorbarcode opened a new pull request, #24580:
URL: https://github.com/apache/pulsar/pull/24580
### Motivation
**Background: how ZK metadata store handles the node exists error when creating a new node**
ZK metadata store throws a `BadVersionException` error if it receives a `NODEEXISTS` error when creating a new node.
https://github.com/apache/pulsar/blob/v4.0.5/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java#L269-L272
```java
if (code == Code.NODEEXISTS) {
// We're emulating a request to create a node, so the version is invalid
op.getFuture().completeExceptionally(getException(Code.BADVERSION, op.getPath()));
}
```
**Background: how ZK metadata store handles disconnect error when executing operations**
ZK metadata store will retry the operation if it receives a `CONNECTIONLOSS` error when executing operations.
https://github.com/apache/pulsar/blob/v4.0.5/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java#L200-L227
```java
if (code == Code.CONNECTIONLOSS) {
// Retry with the individual operations
executor.schedule(() -> {
ops.forEach(o -> batchOperation(Collections.singletonList(o)));
}, 100, TimeUnit.MILLISECONDS);
}
```
**Background: how ZK clients/servers limit the max length of packet**
- ZK clients/servers limit the max length of a packet to 1MB by default.
- Users can use a Java environment variable `jute.maxbuffer` to change the maximum packet length, which controls the maximum packet length for both requests and receives.
- Once the maximum packet length is reached, the ZK client will throw an error and reconnect.
**Issue: ZK client throws an error and reconnects when the max length of the packet is exceeded by the server**
- As the background "How ZK clients/servers limit the max length of packet" said, not only is the request packet is in limiting, but the response packet is not limited.
- The requests that type `Get` or `GET_CHILDREN` will receive a response from the server, and the response packet may be larger than the max length of packet length.
- When the response packet is larger than the maximum packet length, the ZK client will throw an error and reconnect.
- The reconnecting will cause the `BadVersionException` error mentioned in the background "How ZK metadata store handles the node exists error when creating a new node".
- Then the operation of creating a node can never be completed.
You can reproduce the issue by running the new test `testReceivedHugeResponse`
### Modifications
Add a mechanism to limit `get/getChildren` operations of the ZK metadata store. This mechanism merely avoids problems with as few resources as possible, but it cannot prevent all problems. To solve all the problems, there will be a PIP later to make the Settings configurable, see also the section "Next things to do"
### Next things to do
I will submit a PIP
- Make `MetadataNodePayloadLenEstimator` configurable
- Add metrics of the node path that has the max packet length of each type
- Fix the issue of out-of-order command execution in the metadata store. I will describe this problem in detail in PIP
- Solve the following issue
**The Metadata store can not confirm whether the bad version error is expected or not after a reconnection.**
- ZK client sent a `Put` request.
- Server has received the request and is handling it.
- Server is responding.
- The channel reconnected before the `Put` request was responded.
- The `Put` request is sent again, but the node already exists(or already modified), so Pulsar get a `BadVersionException` error.
- **(Highlight)** The previous change is caused by the first try, but the second try is stuck due to the ZK node has been changed by the first try.
- Once this situation happens, we can only restart the broker or reload topics to fix it, since we do not know whether the bad version error is expected or not.
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
### Matching PR in forked repository
PR in forked repository: x
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2240672071
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataNodePayloadLenEstimator.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.metadata.api;
+
+import java.util.List;
+
+public interface MetadataNodePayloadLenEstimator {
+
+ void recordPut(String path, byte[] data);
+
+ void recordGetRes(String path, GetResult getResult);
+
+ void recordGetChildrenRes(String path, List<String> list);
+
+ int estimateGetResPayloadLen(String path);
+
+ int estimateGetChildrenResPayloadLen(String path);
+}
Review Comment:
It would be useful to add javadocs to the interface level and all methods.
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2242159314
##########
pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/MaxValueMetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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.zookeeper;
+
+import io.netty.util.concurrent.FastThreadLocal;
+import java.util.List;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataNodePayloadLenEstimator;
+
+@Slf4j
+public class MaxValueMetadataNodePayloadLenEstimator implements MetadataNodePayloadLenEstimator {
+
+ // Default to max int value, let the first query command do not execute with batch.
+ public static final int DEFAULT_LEN = Integer.MAX_VALUE;
+ public static final int UNSET = -1;
+ // xid: int, zxid: long, err: int, len: int -> 20
+ // Node stat:
+ // czxid: long, mzxid: long, ctime: long, mtime: long, version: int, cversion: int, aversion: int, -> 44
+ // ephemeralOwner: long, dataLength: int, numChildren: int, pzxid: long -> 24
+ // total: 88, let's double it to cover different serialization versions.
+ public static final int ZK_PACKET_SYSTEM_PROPS_LEN = 176;
+ private static final SplitPathRes MEANINGLESS_SPLIT_PATH_RES = new SplitPathRes();
+ private static final FastThreadLocal<SplitPathRes> LOCAL_SPLIT_PATH_RES = new FastThreadLocal<SplitPathRes>() {
+ @Override
+ protected SplitPathRes initialValue() {
+ return new SplitPathRes();
+ }
+ };
+ private final int[] maxLenOfGetMapping;
+ private final int[] maxLenOfListMapping;
+
+ {
+ int pathTypeCount = PathType.values().length;
+ maxLenOfGetMapping = new int[pathTypeCount];
+ maxLenOfListMapping = new int[pathTypeCount];
+ for (int i = 0; i < pathTypeCount; i++) {
+ maxLenOfGetMapping[i] = UNSET;
+ maxLenOfListMapping[i] = UNSET;
+ }
+ }
Review Comment:
could we use ordinary constructor syntax 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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2244606582
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.metadata.api;
+
+import java.util.List;
+
+public interface MetadataNodePayloadLenEstimator {
Review Comment:
Please add the javadoc for the interface itself so that it becomes clear what this is about
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2265950485
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataStoreConfig.java:
##########
@@ -99,4 +99,9 @@ public class MetadataStoreConfig {
*/
@Builder.Default
private OpenTelemetry openTelemetry = OpenTelemetry.noop();
+
+ /**
+ * The estimator to estimate the payload length of metadata node, which used to limit the batch size requested.
Review Comment:
Fix grammar and improvement this javadoc. It's useful to also expand the explanation to mention the ZooKeeper use case so that it's more understandable.
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2242475636
##########
pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/MaxValueMetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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.zookeeper;
+
+import io.netty.util.concurrent.FastThreadLocal;
+import java.util.List;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataNodePayloadLenEstimator;
+
+@Slf4j
+public class MaxValueMetadataNodePayloadLenEstimator implements MetadataNodePayloadLenEstimator {
+
+ // Default to max int value, let the first query command do not execute with batch.
+ public static final int DEFAULT_LEN = Integer.MAX_VALUE;
+ public static final int UNSET = -1;
+ // xid: int, zxid: long, err: int, len: int -> 20
+ // Node stat:
+ // czxid: long, mzxid: long, ctime: long, mtime: long, version: int, cversion: int, aversion: int, -> 44
+ // ephemeralOwner: long, dataLength: int, numChildren: int, pzxid: long -> 24
+ // total: 88, let's double it to cover different serialization versions.
+ public static final int ZK_PACKET_SYSTEM_PROPS_LEN = 176;
+ private static final SplitPathRes MEANINGLESS_SPLIT_PATH_RES = new SplitPathRes();
+ private static final FastThreadLocal<SplitPathRes> LOCAL_SPLIT_PATH_RES = new FastThreadLocal<SplitPathRes>() {
+ @Override
+ protected SplitPathRes initialValue() {
+ return new SplitPathRes();
+ }
+ };
+ private final int[] maxLenOfGetMapping;
+ private final int[] maxLenOfListMapping;
+
+ {
+ int pathTypeCount = PathType.values().length;
+ maxLenOfGetMapping = new int[pathTypeCount];
+ maxLenOfListMapping = new int[pathTypeCount];
+ for (int i = 0; i < pathTypeCount; i++) {
+ maxLenOfGetMapping[i] = UNSET;
+ maxLenOfListMapping[i] = UNSET;
+ }
+ }
Review Comment:
Could you show an example?
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2265953103
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataStoreConfig.java:
##########
@@ -99,4 +99,9 @@ public class MetadataStoreConfig {
*/
@Builder.Default
private OpenTelemetry openTelemetry = OpenTelemetry.noop();
+
+ /**
+ * The estimator to estimate the payload length of metadata node, which used to limit the batch size requested.
Review Comment:
You can also reference the javadoc of MetadataNodePayloadLenEstimator with `{@link MetadataNodePayloadLenEstimator}`
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2241550699
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataNodePayloadLenEstimator.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.metadata.api;
+
+import java.util.List;
+
+public interface MetadataNodePayloadLenEstimator {
+
+ void recordPut(String path, byte[] data);
+
+ void recordGetRes(String path, GetResult getResult);
+
+ void recordGetChildrenRes(String path, List<String> list);
+
+ int estimateGetResPayloadLen(String path);
+
+ int estimateGetChildrenResPayloadLen(String path);
+}
Review Comment:
Added javadocs
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2244590972
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.metadata.api;
+
+import java.util.List;
+
+public interface MetadataNodePayloadLenEstimator {
+
+ /**
+ * Record the payload length of put operation, to let the estimator know the max payload length, which benefits the
+ * accuracy of estimation.
+ */
+ void recordPut(String path, byte[] data);
+
+ /**
+ * Record the payload length of get result, to let the estimator know the max payload length, which benefits the
+ * accuracy of estimation.
+ */
+ void recordGetRes(String path, GetResult getResult);
+
+ /**
+ * Record the payload length of list result, to let the estimator know the max payload length, which benefits the
+ * accuracy of estimation.
+ */
+ void recordGetChildrenRes(String path, List<String> list);
+
+ /**
+ * Estimate the payload length of get result.
+ */
+ int estimateGetResPayloadLen(String path);
+
+ /**
+ * Estimate the payload length of list result.
+ */
+ int estimateGetChildrenResPayloadLen(String path);
Review Comment:
For estimating the size of get children calls, I think that it would be very useful if there would be a way to first retrieve the number of children (ZK has numChildren in stats) to be able to do this calculation. It doesn't make sense to do this estimation for all calls, but for listing topics in a namespace, this would be very useful.
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2265957215
##########
pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/MaxValueMetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,211 @@
+/*
+ * 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.zookeeper;
+
+import io.netty.util.concurrent.FastThreadLocal;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReferenceArray;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataNodePayloadLenEstimator;
+
+@Slf4j
+public class MaxValueMetadataNodePayloadLenEstimator implements MetadataNodePayloadLenEstimator {
+
+ // Default to max int value, let the first query command do not execute with batch.
+ public static final int DEFAULT_LEN = Integer.MAX_VALUE;
+ public static final int UNSET = -1;
+ // xid: int, zxid: long, err: int, len: int -> 20
+ // Node stat:
+ // czxid: long, mzxid: long, ctime: long, mtime: long, version: int, cversion: int, aversion: int, -> 44
+ // ephemeralOwner: long, dataLength: int, numChildren: int, pzxid: long -> 24
+ // total: 88, let's double it to cover different serialization versions.
Review Comment:
please add some explanation to this comment. it's currently not very clear.
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#issuecomment-3177488245
> ZK clients/servers limit the max length of a packet to 1MB by default.
In Pulsar, we have added `-Djute.maxbuffer=10485760"` which change the max length to 10MB. Please help update the PR description to avoid the confusion
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2265938143
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java:
##########
@@ -91,7 +93,12 @@ public abstract class AbstractMetadataStore implements MetadataStoreExtended, Co
protected abstract CompletableFuture<Boolean> existsFromStore(String path);
- protected AbstractMetadataStore(String metadataStoreName, OpenTelemetry openTelemetry) {
+ protected MetadataNodePayloadLenEstimator payloadLenRegistrar;
Review Comment:
why is the field name called `payloadLenRegistrar`?
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2286803027
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataStoreConfig.java:
##########
@@ -99,4 +99,9 @@ public class MetadataStoreConfig {
*/
@Builder.Default
private OpenTelemetry openTelemetry = OpenTelemetry.noop();
+
+ /**
+ * The estimator to estimate the payload length of metadata node, which used to limit the batch size requested.
Review Comment:
I have redefined the interfaces
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java:
##########
@@ -91,7 +93,12 @@ public abstract class AbstractMetadataStore implements MetadataStoreExtended, Co
protected abstract CompletableFuture<Boolean> existsFromStore(String path);
- protected AbstractMetadataStore(String metadataStoreName, OpenTelemetry openTelemetry) {
+ protected MetadataNodePayloadLenEstimator payloadLenRegistrar;
Review Comment:
I have redefined the interfaces
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2242651790
##########
pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/MaxValueMetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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.zookeeper;
+
+import io.netty.util.concurrent.FastThreadLocal;
+import java.util.List;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataNodePayloadLenEstimator;
+
+@Slf4j
+public class MaxValueMetadataNodePayloadLenEstimator implements MetadataNodePayloadLenEstimator {
+
+ // Default to max int value, let the first query command do not execute with batch.
+ public static final int DEFAULT_LEN = Integer.MAX_VALUE;
+ public static final int UNSET = -1;
+ // xid: int, zxid: long, err: int, len: int -> 20
+ // Node stat:
+ // czxid: long, mzxid: long, ctime: long, mtime: long, version: int, cversion: int, aversion: int, -> 44
+ // ephemeralOwner: long, dataLength: int, numChildren: int, pzxid: long -> 24
+ // total: 88, let's double it to cover different serialization versions.
+ public static final int ZK_PACKET_SYSTEM_PROPS_LEN = 176;
+ private static final SplitPathRes MEANINGLESS_SPLIT_PATH_RES = new SplitPathRes();
+ private static final FastThreadLocal<SplitPathRes> LOCAL_SPLIT_PATH_RES = new FastThreadLocal<SplitPathRes>() {
+ @Override
+ protected SplitPathRes initialValue() {
+ return new SplitPathRes();
+ }
+ };
+ private final int[] maxLenOfGetMapping;
+ private final int[] maxLenOfListMapping;
+
+ {
+ int pathTypeCount = PathType.values().length;
+ maxLenOfGetMapping = new int[pathTypeCount];
+ maxLenOfListMapping = new int[pathTypeCount];
+ for (int i = 0; i < pathTypeCount; i++) {
+ maxLenOfGetMapping[i] = UNSET;
+ maxLenOfListMapping[i] = UNSET;
+ }
+ }
Review Comment:
replace the first `{` with `MaxValueMetadataNodePayloadLenEstimator() {` so that it becomes a constructor.
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2240682522
##########
pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/MaxValueMetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.zookeeper;
+
+import java.util.HashMap;
+import java.util.List;
+import org.apache.pulsar.common.util.StringSplitter;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataNodePayloadLenEstimator;
+
+public class MaxValueMetadataNodePayloadLenEstimator implements MetadataNodePayloadLenEstimator {
+
+ private final HashMap<PathType, Integer> maxLenPerPathType = new HashMap<>();
+ private final HashMap<PathType, Integer> maxLenPerListPathType = new HashMap<>();
+ // Default to max int value, let the first query command do not execute with batch.
+ private static final int DEFAULT_LEN = 1;//Integer.MAX_VALUE;
+
+ @Override
+ public void recordPut(String path, byte[] data) {
+ PathType pathType = getPathType(path);
+ maxLenPerPathType.compute(pathType, (k, v) -> {
+ if (v == null) {
+ return data.length;
+ }
+ return Math.max(v, data.length);
+ });
+ }
+
+ @Override
+ public void recordGetRes(String path, GetResult getResult) {
+ PathType pathType = getPathType(path);
+ maxLenPerPathType.compute(pathType, (k, v) -> {
+ if (v == null) {
+ return getResult.getValue().length;
+ }
+ return Math.max(v, getResult.getValue().length);
+ });
+ }
+
+ @Override
+ public void recordGetChildrenRes(String path, List<String> list) {
+ PathType pathType = getPathType(path);
+ maxLenPerListPathType.compute(pathType, (k, v) -> {
+ if (v == null) {
+ return list.stream().map(String::length).reduce(0, Integer::sum);
Review Comment:
```suggestion
return list.stream().mapToInt(String::length).sum();
```
##########
pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/MaxValueMetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.zookeeper;
+
+import java.util.HashMap;
+import java.util.List;
+import org.apache.pulsar.common.util.StringSplitter;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataNodePayloadLenEstimator;
+
+public class MaxValueMetadataNodePayloadLenEstimator implements MetadataNodePayloadLenEstimator {
+
+ private final HashMap<PathType, Integer> maxLenPerPathType = new HashMap<>();
+ private final HashMap<PathType, Integer> maxLenPerListPathType = new HashMap<>();
+ // Default to max int value, let the first query command do not execute with batch.
+ private static final int DEFAULT_LEN = 1;//Integer.MAX_VALUE;
+
+ @Override
+ public void recordPut(String path, byte[] data) {
+ PathType pathType = getPathType(path);
+ maxLenPerPathType.compute(pathType, (k, v) -> {
+ if (v == null) {
+ return data.length;
+ }
+ return Math.max(v, data.length);
+ });
+ }
+
+ @Override
+ public void recordGetRes(String path, GetResult getResult) {
+ PathType pathType = getPathType(path);
+ maxLenPerPathType.compute(pathType, (k, v) -> {
+ if (v == null) {
+ return getResult.getValue().length;
+ }
+ return Math.max(v, getResult.getValue().length);
+ });
+ }
+
+ @Override
+ public void recordGetChildrenRes(String path, List<String> list) {
+ PathType pathType = getPathType(path);
+ maxLenPerListPathType.compute(pathType, (k, v) -> {
+ if (v == null) {
+ return list.stream().map(String::length).reduce(0, Integer::sum);
+ }
+ return Math.max(v, list.stream().map(String::length).reduce(0, Integer::sum));
+ });
+ }
+
+ @Override
+ public int estimateGetResPayloadLen(String path) {
+ PathType pathType = getPathType(path);
+ return maxLenPerPathType.getOrDefault(pathType, DEFAULT_LEN);
+ }
+
+ @Override
+ public int estimateGetChildrenResPayloadLen(String path) {
+ PathType pathType = getPathType(path);
+ int maxValueCached = maxLenPerListPathType.getOrDefault(pathType, DEFAULT_LEN);
+ return maxValueCached + (maxValueCached >> 3);
+ }
+
+ private PathType getPathType(String path) {
+ String[] parts = StringSplitter.splitByChar(path, '/', 6).toArray(new String[0]);
+ if (parts.length < 2) {
+ return PathType.OTHERS;
+ }
+ return switch (parts[0]) {
+ case "admin" -> getAdminPathType(parts);
+ case "managed-ledgers" -> getMlPathType(parts);
+ default -> PathType.OTHERS;
+ };
+ }
+
+ private PathType getAdminPathType(String[] parts) {
+ return switch (parts[1]) {
+ case "clusters" -> PathType.CLUSTER;
+ case "policies" -> switch (parts.length) {
+ case 2 -> PathType.TENANT;
+ case 3 -> PathType.NAMESPACE_POLICIES;
+ default -> PathType.OTHERS;
+ };
+ case "partitioned-topics" -> switch (parts.length) {
+ case 5 -> PathType.PARTITIONED_NAMESPACE;
+ case 6 -> PathType.PARTITIONED_TOPIC;
+ default -> PathType.OTHERS;
+ };
+ default -> PathType.OTHERS;
+ };
+ }
+
+ private PathType getMlPathType(String[] parts) {
+ return switch (parts.length) {
+ case 4 -> PathType.ML_NAMESPACE;
+ case 5 -> PathType.TOPIC;
+ case 6 -> PathType.SUBSCRIPTION;
+ default -> PathType.OTHERS;
+ };
+ }
Review Comment:
not sure if it matters, but V1 topics use a different persistent path than V2 topics.
https://github.com/apache/pulsar/blob/ae050b9c44ef0be474cedcf78c4dd6ba31cf6cc8/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java#L333-L349
##########
pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/MaxValueMetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.zookeeper;
+
+import java.util.HashMap;
+import java.util.List;
+import org.apache.pulsar.common.util.StringSplitter;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataNodePayloadLenEstimator;
+
+public class MaxValueMetadataNodePayloadLenEstimator implements MetadataNodePayloadLenEstimator {
+
+ private final HashMap<PathType, Integer> maxLenPerPathType = new HashMap<>();
+ private final HashMap<PathType, Integer> maxLenPerListPathType = new HashMap<>();
+ // Default to max int value, let the first query command do not execute with batch.
+ private static final int DEFAULT_LEN = 1;//Integer.MAX_VALUE;
+
+ @Override
+ public void recordPut(String path, byte[] data) {
+ PathType pathType = getPathType(path);
+ maxLenPerPathType.compute(pathType, (k, v) -> {
+ if (v == null) {
+ return data.length;
+ }
+ return Math.max(v, data.length);
+ });
+ }
+
+ @Override
+ public void recordGetRes(String path, GetResult getResult) {
+ PathType pathType = getPathType(path);
+ maxLenPerPathType.compute(pathType, (k, v) -> {
+ if (v == null) {
+ return getResult.getValue().length;
+ }
+ return Math.max(v, getResult.getValue().length);
+ });
+ }
+
+ @Override
+ public void recordGetChildrenRes(String path, List<String> list) {
+ PathType pathType = getPathType(path);
+ maxLenPerListPathType.compute(pathType, (k, v) -> {
+ if (v == null) {
+ return list.stream().map(String::length).reduce(0, Integer::sum);
+ }
+ return Math.max(v, list.stream().map(String::length).reduce(0, Integer::sum));
Review Comment:
```suggestion
return Math.max(v, list.stream().mapToInt(String::length).sum());
```
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#issuecomment-3203866319
@lhotari @codelipenghui
I have rewritten with a new solution, could you please review 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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#issuecomment-3279191730
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/24580?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `94.98069%` with `13 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 74.31%. Comparing base ([`cde4948`](https://app.codecov.io/gh/apache/pulsar/commit/cde4948044767033e40bc7d4eff98a4c25fd3aff?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`511859f`](https://app.codecov.io/gh/apache/pulsar/commit/511859fddbc088ba186ad5f191d24ea9bbf1dc13?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 64 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24580?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...pulsar/zookeeper/DefaultMetadataNodeSizeStats.java](https://app.codecov.io/gh/apache/pulsar/pull/24580?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fzookeeper%2FDefaultMetadataNodeSizeStats.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL3pvb2tlZXBlci9EZWZhdWx0TWV0YWRhdGFOb2RlU2l6ZVN0YXRzLmphdmE=) | 97.82% | [1 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24580?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...g/apache/pulsar/metadata/impl/ZKMetadataStore.java](https://app.codecov.io/gh/apache/pulsar/pull/24580?src=pr&el=tree&filepath=pulsar-metadata%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fmetadata%2Fimpl%2FZKMetadataStore.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLW1ldGFkYXRhL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9wdWxzYXIvbWV0YWRhdGEvaW1wbC9aS01ldGFkYXRhU3RvcmUuamF2YQ==) | 84.21% | [0 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24580?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ta/impl/batching/ZKMetadataStoreBatchStrategy.java](https://app.codecov.io/gh/apache/pulsar/pull/24580?src=pr&el=tree&filepath=pulsar-metadata%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fmetadata%2Fimpl%2Fbatching%2FZKMetadataStoreBatchStrategy.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLW1ldGFkYXRhL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9wdWxzYXIvbWV0YWRhdGEvaW1wbC9iYXRjaGluZy9aS01ldGFkYXRhU3RvcmVCYXRjaFN0cmF0ZWd5LmphdmE=) | 94.00% | [2 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24580?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...pl/batching/DefaultMetadataStoreBatchStrategy.java](https://app.codecov.io/gh/apache/pulsar/pull/24580?src=pr&el=tree&filepath=pulsar-metadata%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fmetadata%2Fimpl%2Fbatching%2FDefaultMetadataStoreBatchStrategy.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLW1ldGFkYXRhL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9wdWxzYXIvbWV0YWRhdGEvaW1wbC9iYXRjaGluZy9EZWZhdWx0TWV0YWRhdGFTdG9yZUJhdGNoU3RyYXRlZ3kuamF2YQ==) | 90.47% | [1 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24580?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ulsar/metadata/api/DummyMetadataNodeSizeStats.java](https://app.codecov.io/gh/apache/pulsar/pull/24580?src=pr&el=tree&filepath=pulsar-metadata%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fmetadata%2Fapi%2FDummyMetadataNodeSizeStats.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLW1ldGFkYXRhL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9wdWxzYXIvbWV0YWRhdGEvYXBpL0R1bW15TWV0YWRhdGFOb2RlU2l6ZVN0YXRzLmphdmE=) | 83.33% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24580?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...e/pulsar/metadata/impl/oxia/OxiaMetadataStore.java](https://app.codecov.io/gh/apache/pulsar/pull/24580?src=pr&el=tree&filepath=pulsar-metadata%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fmetadata%2Fimpl%2Foxia%2FOxiaMetadataStore.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLW1ldGFkYXRhL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9wdWxzYXIvbWV0YWRhdGEvaW1wbC9veGlhL094aWFNZXRhZGF0YVN0b3JlLmphdmE=) | 66.66% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24580?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
[](https://app.codecov.io/gh/apache/pulsar/pull/24580?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #24580 +/- ##
=============================================
+ Coverage 36.00% 74.31% +38.31%
- Complexity 12799 33628 +20829
=============================================
Files 1825 1900 +75
Lines 142779 148339 +5560
Branches 16393 17200 +807
=============================================
+ Hits 51402 110233 +58831
+ Misses 84275 29336 -54939
- Partials 7102 8770 +1668
```
| [Flag](https://app.codecov.io/gh/apache/pulsar/pull/24580/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [inttests](https://app.codecov.io/gh/apache/pulsar/pull/24580/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `26.39% <74.90%> (?)` | |
| [systests](https://app.codecov.io/gh/apache/pulsar/pull/24580/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `22.77% <70.65%> (-0.55%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pulsar/pull/24580/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.82% <94.98%> (+39.39%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24580?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...n/java/org/apache/pulsar/broker/PulsarService.java](https://app.codecov.io/gh/apache/pulsar/pull/24580?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2FPulsarService.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9QdWxzYXJTZXJ2aWNlLmphdmE=) | `84.26% <100.00%> (+17.05%)` | :arrow_up: |
| [...he/pulsar/metadata/impl/AbstractMetadataStore.java](https://app.codecov.io/gh/apache/pulsar/pull/24580?src=pr&el=tree&filepath=pulsar-metadata%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fmetadata%2Fimpl%2FAbstractMetadataStore.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLW1ldGFkYXRhL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9wdWxzYXIvbWV0YWRhdGEvaW1wbC9BYnN0cmFjdE1ldGFkYXRhU3RvcmUuamF2YQ==) | `84.98% <100.00%> (+34.45%)` | :arrow_up: |
| [...pulsar/metadata/impl/LocalMemoryMetadataStore.java](https://app.codecov.io/gh/apache/pulsar/pull/24580?src=pr&el=tree&filepath=pulsar-metadata%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fmetadata%2Fimpl%2FLocalMemoryMetadataStore.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLW1ldGFkYXRhL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9wdWxzYXIvbWV0YWRhdGEvaW1wbC9Mb2NhbE1lbW9yeU1ldGFkYXRhU3RvcmUuamF2YQ==) | `92.30% <100.00%> (+38.90%)` | :arrow_up: |
| [...che/pulsar/metadata/impl/RocksdbMetadataStore.java](https://app.codecov.io/gh/apache/pulsar/pull/24580?src=pr&el=tree&filepath=pulsar-metadata%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fmetadata%2Fimpl%2FRocksdbMetadataStore.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLW1ldGFkYXRhL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9wdWxzYXIvbWV0YWRhdGEvaW1wbC9Sb2Nrc2RiTWV0YWRhdGFTdG9yZS5qYXZh) | `70.94% <100.00%> (+70.60%)` | :arrow_up: |
| [...ta/impl/batching/AbstractBatchedMetadataStore.java](https://app.codecov.io/gh/apache/pulsar/pull/24580?src=pr&el=tree&filepath=pulsar-metadata%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fmetadata%2Fimpl%2Fbatching%2FAbstractBatchedMetadataStore.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLW1ldGFkYXRhL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9wdWxzYXIvbWV0YWRhdGEvaW1wbC9iYXRjaGluZy9BYnN0cmFjdEJhdGNoZWRNZXRhZGF0YVN0b3JlLmphdmE=) | `91.02% <100.00%> (+21.13%)` | :arrow_up: |
| [...ulsar/metadata/api/DummyMetadataNodeSizeStats.java](https://app.codecov.io/gh/apache/pulsar/pull/24580?src=pr&el=tree&filepath=pulsar-metadata%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fmetadata%2Fapi%2FDummyMetadataNodeSizeStats.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLW1ldGFkYXRhL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9wdWxzYXIvbWV0YWRhdGEvYXBpL0R1bW15TWV0YWRhdGFOb2RlU2l6ZVN0YXRzLmphdmE=) | `83.33% <83.33%> (ø)` | |
| [...e/pulsar/metadata/impl/oxia/OxiaMetadataStore.java](https://app.codecov.io/gh/apache/pulsar/pull/24580?src=pr&el=tree&filepath=pulsar-metadata%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fmetadata%2Fimpl%2Foxia%2FOxiaMetadataStore.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLW1ldGFkYXRhL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9wdWxzYXIvbWV0YWRhdGEvaW1wbC9veGlhL094aWFNZXRhZGF0YVN0b3JlLmphdmE=) | `87.91% <66.66%> (+87.91%)` | :arrow_up: |
| [...pl/batching/DefaultMetadataStoreBatchStrategy.java](https://app.codecov.io/gh/apache/pulsar/pull/24580?src=pr&el=tree&filepath=pulsar-metadata%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fmetadata%2Fimpl%2Fbatching%2FDefaultMetadataStoreBatchStrategy.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLW1ldGFkYXRhL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9wdWxzYXIvbWV0YWRhdGEvaW1wbC9iYXRjaGluZy9EZWZhdWx0TWV0YWRhdGFTdG9yZUJhdGNoU3RyYXRlZ3kuamF2YQ==) | `90.47% <90.47%> (ø)` | |
| [...pulsar/zookeeper/DefaultMetadataNodeSizeStats.java](https://app.codecov.io/gh/apache/pulsar/pull/24580?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fzookeeper%2FDefaultMetadataNodeSizeStats.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL3pvb2tlZXBlci9EZWZhdWx0TWV0YWRhdGFOb2RlU2l6ZVN0YXRzLmphdmE=) | `97.82% <97.82%> (ø)` | |
| [...g/apache/pulsar/metadata/impl/ZKMetadataStore.java](https://app.codecov.io/gh/apache/pulsar/pull/24580?src=pr&el=tree&filepath=pulsar-metadata%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fmetadata%2Fimpl%2FZKMetadataStore.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLW1ldGFkYXRhL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9wdWxzYXIvbWV0YWRhdGEvaW1wbC9aS01ldGFkYXRhU3RvcmUuamF2YQ==) | `86.95% <84.21%> (+24.53%)` | :arrow_up: |
| ... and [1 more](https://app.codecov.io/gh/apache/pulsar/pull/24580?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | |
... and [1437 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/24580/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#issuecomment-3135323510
/pulsarbot rerun-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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2247619386
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ZKMetadataStoreBatchIOperationTest.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.broker.service;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble;
+import org.apache.pulsar.zookeeper.ZookeeperServerTest;
+import org.apache.zookeeper.client.ZKClientConfig;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Slf4j
+@Test(groups = "broker")
+public class ZKMetadataStoreBatchIOperationTest extends CanReconnectZKClientPulsarServiceBaseTest {
+
+ @Override
+ @BeforeClass(alwaysRun = true, timeOut = 300000)
+ public void setup() throws Exception {
+ System.setProperty("jute.maxbuffer", "16384");
Review Comment:
Fixed
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.metadata.api;
+
+import java.util.List;
+
+public interface MetadataNodePayloadLenEstimator {
Review Comment:
Added
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2247608353
##########
pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/MaxValueMetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.zookeeper;
+
+import java.util.HashMap;
+import java.util.List;
+import org.apache.pulsar.common.util.StringSplitter;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataNodePayloadLenEstimator;
+
+public class MaxValueMetadataNodePayloadLenEstimator implements MetadataNodePayloadLenEstimator {
+
+ private final HashMap<PathType, Integer> maxLenPerPathType = new HashMap<>();
+ private final HashMap<PathType, Integer> maxLenPerListPathType = new HashMap<>();
Review Comment:
Changed its type to `AtomicReferenceArray`
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2244629461
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ZKMetadataStoreBatchIOperationTest.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.broker.service;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble;
+import org.apache.pulsar.zookeeper.ZookeeperServerTest;
+import org.apache.zookeeper.client.ZKClientConfig;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Slf4j
+@Test(groups = "broker")
+public class ZKMetadataStoreBatchIOperationTest extends CanReconnectZKClientPulsarServiceBaseTest {
+
+ @Override
+ @BeforeClass(alwaysRun = true, timeOut = 300000)
+ public void setup() throws Exception {
+ System.setProperty("jute.maxbuffer", "16384");
Review Comment:
Since this is modified, it should be resetted in cleanup so that other tests aren't impacted.
It might also be necessary to configure the CI run to run this test class in isolation. That would be needed if the ZK client caches the jute.maxbuffer setting and we cannot find a way to reset the cached value with reflection.
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2241540087
##########
pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/MaxValueMetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.zookeeper;
+
+import java.util.HashMap;
+import java.util.List;
+import org.apache.pulsar.common.util.StringSplitter;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataNodePayloadLenEstimator;
+
+public class MaxValueMetadataNodePayloadLenEstimator implements MetadataNodePayloadLenEstimator {
+
+ private final HashMap<PathType, Integer> maxLenPerPathType = new HashMap<>();
+ private final HashMap<PathType, Integer> maxLenPerListPathType = new HashMap<>();
+ // Default to max int value, let the first query command do not execute with batch.
+ private static final int DEFAULT_LEN = 1;//Integer.MAX_VALUE;
+
+ @Override
+ public void recordPut(String path, byte[] data) {
+ PathType pathType = getPathType(path);
+ maxLenPerPathType.compute(pathType, (k, v) -> {
+ if (v == null) {
+ return data.length;
+ }
+ return Math.max(v, data.length);
+ });
+ }
+
+ @Override
+ public void recordGetRes(String path, GetResult getResult) {
+ PathType pathType = getPathType(path);
+ maxLenPerPathType.compute(pathType, (k, v) -> {
+ if (v == null) {
+ return getResult.getValue().length;
+ }
+ return Math.max(v, getResult.getValue().length);
+ });
+ }
+
+ @Override
+ public void recordGetChildrenRes(String path, List<String> list) {
+ PathType pathType = getPathType(path);
+ maxLenPerListPathType.compute(pathType, (k, v) -> {
+ if (v == null) {
+ return list.stream().map(String::length).reduce(0, Integer::sum);
+ }
+ return Math.max(v, list.stream().map(String::length).reduce(0, Integer::sum));
+ });
+ }
+
+ @Override
+ public int estimateGetResPayloadLen(String path) {
+ PathType pathType = getPathType(path);
+ return maxLenPerPathType.getOrDefault(pathType, DEFAULT_LEN);
+ }
+
+ @Override
+ public int estimateGetChildrenResPayloadLen(String path) {
+ PathType pathType = getPathType(path);
+ int maxValueCached = maxLenPerListPathType.getOrDefault(pathType, DEFAULT_LEN);
+ return maxValueCached + (maxValueCached >> 3);
Review Comment:
Removed it and added a more precise method to cover the length of ZK system props in the response
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2240678110
##########
pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/MaxValueMetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.zookeeper;
+
+import java.util.HashMap;
+import java.util.List;
+import org.apache.pulsar.common.util.StringSplitter;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataNodePayloadLenEstimator;
+
+public class MaxValueMetadataNodePayloadLenEstimator implements MetadataNodePayloadLenEstimator {
+
+ private final HashMap<PathType, Integer> maxLenPerPathType = new HashMap<>();
+ private final HashMap<PathType, Integer> maxLenPerListPathType = new HashMap<>();
Review Comment:
HashMap isn't thread safe. It's better to use ConcurrentHashMap to avoid problems.
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2286802765
##########
pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/MaxValueMetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,211 @@
+/*
+ * 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.zookeeper;
+
+import io.netty.util.concurrent.FastThreadLocal;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReferenceArray;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataNodePayloadLenEstimator;
+
+@Slf4j
+public class MaxValueMetadataNodePayloadLenEstimator implements MetadataNodePayloadLenEstimator {
+
+ // Default to max int value, let the first query command do not execute with batch.
+ public static final int DEFAULT_LEN = Integer.MAX_VALUE;
+ public static final int UNSET = -1;
+ // xid: int, zxid: long, err: int, len: int -> 20
+ // Node stat:
+ // czxid: long, mzxid: long, ctime: long, mtime: long, version: int, cversion: int, aversion: int, -> 44
+ // ephemeralOwner: long, dataLength: int, numChildren: int, pzxid: long -> 24
+ // total: 88, let's double it to cover different serialization versions.
+ public static final int ZK_PACKET_SYSTEM_PROPS_LEN = 176;
+ private static final SplitPathRes MEANINGLESS_SPLIT_PATH_RES = new SplitPathRes();
+ private static final FastThreadLocal<SplitPathRes> LOCAL_SPLIT_PATH_RES = new FastThreadLocal<SplitPathRes>() {
+ @Override
+ protected SplitPathRes initialValue() {
+ return new SplitPathRes();
+ }
+ };
+ private final AtomicReferenceArray<Integer> maxLenOfGetMapping;
+ private final AtomicReferenceArray<Integer> maxLenOfListMapping;
+
+ public MaxValueMetadataNodePayloadLenEstimator() {
+ int pathTypeCount = PathType.values().length;
+ maxLenOfGetMapping = new AtomicReferenceArray<>(pathTypeCount);
+ maxLenOfListMapping = new AtomicReferenceArray<>(pathTypeCount);
+ for (int i = 0; i < pathTypeCount; i++) {
+ maxLenOfGetMapping.set(i, UNSET);
+ maxLenOfListMapping.set(i, UNSET);
+ }
+ }
+
+ @Override
+ public void recordPut(String path, byte[] data) {
+ PathType pathType = getPathType(path);
+ maxLenOfGetMapping.set(pathType.ordinal(), Math.max(maxLenOfGetMapping.get(pathType.ordinal()), data.length));
+ }
+
+ @Override
+ public void recordGetRes(String path, GetResult getResult) {
+ PathType pathType = getPathType(path);
+ if (getResult == null) {
+ return;
+ }
+ maxLenOfGetMapping.set(pathType.ordinal(), Math.max(maxLenOfGetMapping.get(pathType.ordinal()),
+ getResult.getValue().length));
+ }
+
+ @Override
+ public void recordGetChildrenRes(String path, List<String> list) {
+ PathType pathType = getPathType(path);
+ // ZK serialize each string with 4 bytes length prefix, so we add 4 bytes for each string.
+ int totalLen = list.stream().mapToInt(String::length).sum() + list.size() * 4;
+ maxLenOfListMapping.set(pathType.ordinal(), Math.max(maxLenOfListMapping.get(pathType.ordinal()), totalLen));
+ }
+
+ @Override
+ public int estimateGetResPayloadLen(String path) {
+ return internalEstimateGetResPayloadLen(path) + ZK_PACKET_SYSTEM_PROPS_LEN;
+ }
+
+ @Override
+ public int estimateGetChildrenResPayloadLen(String path) {
+ return internalEstimateGetChildrenResPayloadLen(path) + ZK_PACKET_SYSTEM_PROPS_LEN;
+ }
+
+ public int internalEstimateGetResPayloadLen(String path) {
+ PathType pathType = getPathType(path);
+ int res = maxLenOfGetMapping.get(pathType.ordinal());
+ return res == UNSET ? DEFAULT_LEN : res;
+ }
+
+ public int internalEstimateGetChildrenResPayloadLen(String path) {
+ PathType pathType = getPathType(path);
+ int res = maxLenOfListMapping.get(pathType.ordinal());
+ return res == UNSET ? DEFAULT_LEN : res;
+ }
+
+ private PathType getPathType(String path) {
+ SplitPathRes splitPathRes = splitPath(path);
+ if (splitPathRes.partCount < 2) {
+ return PathType.OTHERS;
+ }
+ return switch (splitPathRes.parts[0]) {
+ case "admin" -> getAdminPathType(splitPathRes);
+ case "managed-ledgers" -> getMlPathType(splitPathRes);
+ default -> PathType.OTHERS;
+ };
+ }
Review Comment:
I have redefined the interfaces
##########
pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/MaxValueMetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,211 @@
+/*
+ * 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.zookeeper;
+
+import io.netty.util.concurrent.FastThreadLocal;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReferenceArray;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataNodePayloadLenEstimator;
+
+@Slf4j
+public class MaxValueMetadataNodePayloadLenEstimator implements MetadataNodePayloadLenEstimator {
+
+ // Default to max int value, let the first query command do not execute with batch.
+ public static final int DEFAULT_LEN = Integer.MAX_VALUE;
+ public static final int UNSET = -1;
+ // xid: int, zxid: long, err: int, len: int -> 20
+ // Node stat:
+ // czxid: long, mzxid: long, ctime: long, mtime: long, version: int, cversion: int, aversion: int, -> 44
+ // ephemeralOwner: long, dataLength: int, numChildren: int, pzxid: long -> 24
+ // total: 88, let's double it to cover different serialization versions.
+ public static final int ZK_PACKET_SYSTEM_PROPS_LEN = 176;
+ private static final SplitPathRes MEANINGLESS_SPLIT_PATH_RES = new SplitPathRes();
+ private static final FastThreadLocal<SplitPathRes> LOCAL_SPLIT_PATH_RES = new FastThreadLocal<SplitPathRes>() {
+ @Override
+ protected SplitPathRes initialValue() {
+ return new SplitPathRes();
+ }
+ };
+ private final AtomicReferenceArray<Integer> maxLenOfGetMapping;
+ private final AtomicReferenceArray<Integer> maxLenOfListMapping;
+
+ public MaxValueMetadataNodePayloadLenEstimator() {
+ int pathTypeCount = PathType.values().length;
+ maxLenOfGetMapping = new AtomicReferenceArray<>(pathTypeCount);
+ maxLenOfListMapping = new AtomicReferenceArray<>(pathTypeCount);
+ for (int i = 0; i < pathTypeCount; i++) {
+ maxLenOfGetMapping.set(i, UNSET);
+ maxLenOfListMapping.set(i, UNSET);
+ }
+ }
+
+ @Override
+ public void recordPut(String path, byte[] data) {
+ PathType pathType = getPathType(path);
+ maxLenOfGetMapping.set(pathType.ordinal(), Math.max(maxLenOfGetMapping.get(pathType.ordinal()), data.length));
+ }
+
+ @Override
+ public void recordGetRes(String path, GetResult getResult) {
+ PathType pathType = getPathType(path);
+ if (getResult == null) {
+ return;
+ }
+ maxLenOfGetMapping.set(pathType.ordinal(), Math.max(maxLenOfGetMapping.get(pathType.ordinal()),
+ getResult.getValue().length));
+ }
+
+ @Override
+ public void recordGetChildrenRes(String path, List<String> list) {
+ PathType pathType = getPathType(path);
+ // ZK serialize each string with 4 bytes length prefix, so we add 4 bytes for each string.
+ int totalLen = list.stream().mapToInt(String::length).sum() + list.size() * 4;
+ maxLenOfListMapping.set(pathType.ordinal(), Math.max(maxLenOfListMapping.get(pathType.ordinal()), totalLen));
+ }
+
+ @Override
+ public int estimateGetResPayloadLen(String path) {
+ return internalEstimateGetResPayloadLen(path) + ZK_PACKET_SYSTEM_PROPS_LEN;
Review Comment:
I have redefined the interfaces
##########
pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/MaxValueMetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,211 @@
+/*
+ * 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.zookeeper;
+
+import io.netty.util.concurrent.FastThreadLocal;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReferenceArray;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataNodePayloadLenEstimator;
+
+@Slf4j
+public class MaxValueMetadataNodePayloadLenEstimator implements MetadataNodePayloadLenEstimator {
+
+ // Default to max int value, let the first query command do not execute with batch.
+ public static final int DEFAULT_LEN = Integer.MAX_VALUE;
+ public static final int UNSET = -1;
+ // xid: int, zxid: long, err: int, len: int -> 20
+ // Node stat:
+ // czxid: long, mzxid: long, ctime: long, mtime: long, version: int, cversion: int, aversion: int, -> 44
+ // ephemeralOwner: long, dataLength: int, numChildren: int, pzxid: long -> 24
+ // total: 88, let's double it to cover different serialization versions.
Review Comment:
I have redefined the interfaces
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode merged PR #24580:
URL: https://github.com/apache/pulsar/pull/24580
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2241538200
##########
pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/MaxValueMetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.zookeeper;
+
+import java.util.HashMap;
+import java.util.List;
+import org.apache.pulsar.common.util.StringSplitter;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataNodePayloadLenEstimator;
+
+public class MaxValueMetadataNodePayloadLenEstimator implements MetadataNodePayloadLenEstimator {
+
+ private final HashMap<PathType, Integer> maxLenPerPathType = new HashMap<>();
+ private final HashMap<PathType, Integer> maxLenPerListPathType = new HashMap<>();
+ // Default to max int value, let the first query command do not execute with batch.
+ private static final int DEFAULT_LEN = 1;//Integer.MAX_VALUE;
+
+ @Override
+ public void recordPut(String path, byte[] data) {
+ PathType pathType = getPathType(path);
+ maxLenPerPathType.compute(pathType, (k, v) -> {
+ if (v == null) {
+ return data.length;
+ }
+ return Math.max(v, data.length);
+ });
+ }
+
+ @Override
+ public void recordGetRes(String path, GetResult getResult) {
+ PathType pathType = getPathType(path);
+ maxLenPerPathType.compute(pathType, (k, v) -> {
+ if (v == null) {
+ return getResult.getValue().length;
+ }
+ return Math.max(v, getResult.getValue().length);
+ });
+ }
+
+ @Override
+ public void recordGetChildrenRes(String path, List<String> list) {
+ PathType pathType = getPathType(path);
+ maxLenPerListPathType.compute(pathType, (k, v) -> {
+ if (v == null) {
+ return list.stream().map(String::length).reduce(0, Integer::sum);
+ }
+ return Math.max(v, list.stream().map(String::length).reduce(0, Integer::sum));
+ });
+ }
+
+ @Override
+ public int estimateGetResPayloadLen(String path) {
+ PathType pathType = getPathType(path);
+ return maxLenPerPathType.getOrDefault(pathType, DEFAULT_LEN);
+ }
+
+ @Override
+ public int estimateGetChildrenResPayloadLen(String path) {
+ PathType pathType = getPathType(path);
+ int maxValueCached = maxLenPerListPathType.getOrDefault(pathType, DEFAULT_LEN);
+ return maxValueCached + (maxValueCached >> 3);
+ }
+
+ private PathType getPathType(String path) {
+ String[] parts = StringSplitter.splitByChar(path, '/', 6).toArray(new String[0]);
+ if (parts.length < 2) {
+ return PathType.OTHERS;
+ }
+ return switch (parts[0]) {
+ case "admin" -> getAdminPathType(parts);
+ case "managed-ledgers" -> getMlPathType(parts);
+ default -> PathType.OTHERS;
+ };
+ }
+
+ private PathType getAdminPathType(String[] parts) {
+ return switch (parts[1]) {
+ case "clusters" -> PathType.CLUSTER;
+ case "policies" -> switch (parts.length) {
+ case 2 -> PathType.TENANT;
+ case 3 -> PathType.NAMESPACE_POLICIES;
+ default -> PathType.OTHERS;
+ };
+ case "partitioned-topics" -> switch (parts.length) {
+ case 5 -> PathType.PARTITIONED_NAMESPACE;
+ case 6 -> PathType.PARTITIONED_TOPIC;
+ default -> PathType.OTHERS;
+ };
+ default -> PathType.OTHERS;
+ };
+ }
+
+ private PathType getMlPathType(String[] parts) {
+ return switch (parts.length) {
+ case 4 -> PathType.ML_NAMESPACE;
+ case 5 -> PathType.TOPIC;
+ case 6 -> PathType.SUBSCRIPTION;
+ default -> PathType.OTHERS;
+ };
+ }
Review Comment:
V1 topic will be calculated as subscription, it is okay.
I added a new case handling for the v1 subscription.
```java
// v2 subscription and v1 topic.
case 6 -> PathType.SUBSCRIPTION;
// v1 subscription.
case 7 -> PathType.SUBSCRIPTION;
```
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2241543928
##########
pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/MaxValueMetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.zookeeper;
+
+import java.util.HashMap;
+import java.util.List;
+import org.apache.pulsar.common.util.StringSplitter;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataNodePayloadLenEstimator;
+
+public class MaxValueMetadataNodePayloadLenEstimator implements MetadataNodePayloadLenEstimator {
+
+ private final HashMap<PathType, Integer> maxLenPerPathType = new HashMap<>();
+ private final HashMap<PathType, Integer> maxLenPerListPathType = new HashMap<>();
Review Comment:
I have used an array instead of the map, which will not be expanded or shrunk anymore. It is not important enough to ensure thread safety for an accurate value, so skipped to guarantee the thread safety
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2265963535
##########
pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/MaxValueMetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,211 @@
+/*
+ * 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.zookeeper;
+
+import io.netty.util.concurrent.FastThreadLocal;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReferenceArray;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataNodePayloadLenEstimator;
+
+@Slf4j
+public class MaxValueMetadataNodePayloadLenEstimator implements MetadataNodePayloadLenEstimator {
+
+ // Default to max int value, let the first query command do not execute with batch.
+ public static final int DEFAULT_LEN = Integer.MAX_VALUE;
+ public static final int UNSET = -1;
+ // xid: int, zxid: long, err: int, len: int -> 20
+ // Node stat:
+ // czxid: long, mzxid: long, ctime: long, mtime: long, version: int, cversion: int, aversion: int, -> 44
+ // ephemeralOwner: long, dataLength: int, numChildren: int, pzxid: long -> 24
+ // total: 88, let's double it to cover different serialization versions.
+ public static final int ZK_PACKET_SYSTEM_PROPS_LEN = 176;
+ private static final SplitPathRes MEANINGLESS_SPLIT_PATH_RES = new SplitPathRes();
+ private static final FastThreadLocal<SplitPathRes> LOCAL_SPLIT_PATH_RES = new FastThreadLocal<SplitPathRes>() {
+ @Override
+ protected SplitPathRes initialValue() {
+ return new SplitPathRes();
+ }
+ };
+ private final AtomicReferenceArray<Integer> maxLenOfGetMapping;
+ private final AtomicReferenceArray<Integer> maxLenOfListMapping;
+
+ public MaxValueMetadataNodePayloadLenEstimator() {
+ int pathTypeCount = PathType.values().length;
+ maxLenOfGetMapping = new AtomicReferenceArray<>(pathTypeCount);
+ maxLenOfListMapping = new AtomicReferenceArray<>(pathTypeCount);
+ for (int i = 0; i < pathTypeCount; i++) {
+ maxLenOfGetMapping.set(i, UNSET);
+ maxLenOfListMapping.set(i, UNSET);
+ }
+ }
+
+ @Override
+ public void recordPut(String path, byte[] data) {
+ PathType pathType = getPathType(path);
+ maxLenOfGetMapping.set(pathType.ordinal(), Math.max(maxLenOfGetMapping.get(pathType.ordinal()), data.length));
+ }
+
+ @Override
+ public void recordGetRes(String path, GetResult getResult) {
+ PathType pathType = getPathType(path);
+ if (getResult == null) {
+ return;
+ }
+ maxLenOfGetMapping.set(pathType.ordinal(), Math.max(maxLenOfGetMapping.get(pathType.ordinal()),
+ getResult.getValue().length));
+ }
+
+ @Override
+ public void recordGetChildrenRes(String path, List<String> list) {
+ PathType pathType = getPathType(path);
+ // ZK serialize each string with 4 bytes length prefix, so we add 4 bytes for each string.
+ int totalLen = list.stream().mapToInt(String::length).sum() + list.size() * 4;
+ maxLenOfListMapping.set(pathType.ordinal(), Math.max(maxLenOfListMapping.get(pathType.ordinal()), totalLen));
+ }
+
+ @Override
+ public int estimateGetResPayloadLen(String path) {
+ return internalEstimateGetResPayloadLen(path) + ZK_PACKET_SYSTEM_PROPS_LEN;
+ }
+
+ @Override
+ public int estimateGetChildrenResPayloadLen(String path) {
+ return internalEstimateGetChildrenResPayloadLen(path) + ZK_PACKET_SYSTEM_PROPS_LEN;
+ }
+
+ public int internalEstimateGetResPayloadLen(String path) {
+ PathType pathType = getPathType(path);
+ int res = maxLenOfGetMapping.get(pathType.ordinal());
+ return res == UNSET ? DEFAULT_LEN : res;
+ }
+
+ public int internalEstimateGetChildrenResPayloadLen(String path) {
+ PathType pathType = getPathType(path);
+ int res = maxLenOfListMapping.get(pathType.ordinal());
+ return res == UNSET ? DEFAULT_LEN : res;
+ }
+
+ private PathType getPathType(String path) {
+ SplitPathRes splitPathRes = splitPath(path);
+ if (splitPathRes.partCount < 2) {
+ return PathType.OTHERS;
+ }
+ return switch (splitPathRes.parts[0]) {
+ case "admin" -> getAdminPathType(splitPathRes);
+ case "managed-ledgers" -> getMlPathType(splitPathRes);
+ default -> PathType.OTHERS;
+ };
+ }
Review Comment:
Does this also work for cases where the ZK prefix (root path) would contain multiple levels?
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2265959613
##########
pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/MaxValueMetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,211 @@
+/*
+ * 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.zookeeper;
+
+import io.netty.util.concurrent.FastThreadLocal;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReferenceArray;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataNodePayloadLenEstimator;
+
+@Slf4j
+public class MaxValueMetadataNodePayloadLenEstimator implements MetadataNodePayloadLenEstimator {
+
+ // Default to max int value, let the first query command do not execute with batch.
+ public static final int DEFAULT_LEN = Integer.MAX_VALUE;
+ public static final int UNSET = -1;
+ // xid: int, zxid: long, err: int, len: int -> 20
+ // Node stat:
+ // czxid: long, mzxid: long, ctime: long, mtime: long, version: int, cversion: int, aversion: int, -> 44
+ // ephemeralOwner: long, dataLength: int, numChildren: int, pzxid: long -> 24
+ // total: 88, let's double it to cover different serialization versions.
+ public static final int ZK_PACKET_SYSTEM_PROPS_LEN = 176;
+ private static final SplitPathRes MEANINGLESS_SPLIT_PATH_RES = new SplitPathRes();
+ private static final FastThreadLocal<SplitPathRes> LOCAL_SPLIT_PATH_RES = new FastThreadLocal<SplitPathRes>() {
+ @Override
+ protected SplitPathRes initialValue() {
+ return new SplitPathRes();
+ }
+ };
+ private final AtomicReferenceArray<Integer> maxLenOfGetMapping;
+ private final AtomicReferenceArray<Integer> maxLenOfListMapping;
+
+ public MaxValueMetadataNodePayloadLenEstimator() {
+ int pathTypeCount = PathType.values().length;
+ maxLenOfGetMapping = new AtomicReferenceArray<>(pathTypeCount);
+ maxLenOfListMapping = new AtomicReferenceArray<>(pathTypeCount);
+ for (int i = 0; i < pathTypeCount; i++) {
+ maxLenOfGetMapping.set(i, UNSET);
+ maxLenOfListMapping.set(i, UNSET);
+ }
+ }
+
+ @Override
+ public void recordPut(String path, byte[] data) {
+ PathType pathType = getPathType(path);
+ maxLenOfGetMapping.set(pathType.ordinal(), Math.max(maxLenOfGetMapping.get(pathType.ordinal()), data.length));
+ }
+
+ @Override
+ public void recordGetRes(String path, GetResult getResult) {
+ PathType pathType = getPathType(path);
+ if (getResult == null) {
+ return;
+ }
+ maxLenOfGetMapping.set(pathType.ordinal(), Math.max(maxLenOfGetMapping.get(pathType.ordinal()),
+ getResult.getValue().length));
+ }
+
+ @Override
+ public void recordGetChildrenRes(String path, List<String> list) {
+ PathType pathType = getPathType(path);
+ // ZK serialize each string with 4 bytes length prefix, so we add 4 bytes for each string.
+ int totalLen = list.stream().mapToInt(String::length).sum() + list.size() * 4;
+ maxLenOfListMapping.set(pathType.ordinal(), Math.max(maxLenOfListMapping.get(pathType.ordinal()), totalLen));
+ }
+
+ @Override
+ public int estimateGetResPayloadLen(String path) {
+ return internalEstimateGetResPayloadLen(path) + ZK_PACKET_SYSTEM_PROPS_LEN;
Review Comment:
why is `ZK_PACKET_SYSTEM_PROPS_LEN` added even in cases where ZK wouldn't be used? Adding a comment could be useful. Is this implementation only for ZooKeeper?
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2265951361
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataStoreConfig.java:
##########
@@ -99,4 +99,9 @@ public class MetadataStoreConfig {
*/
@Builder.Default
private OpenTelemetry openTelemetry = OpenTelemetry.noop();
+
+ /**
+ * The estimator to estimate the payload length of metadata node, which used to limit the batch size requested.
Review Comment:
LLM can be used to check and fix the grammar.
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2242171398
##########
pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/MaxValueMetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.zookeeper;
+
+import java.util.HashMap;
+import java.util.List;
+import org.apache.pulsar.common.util.StringSplitter;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataNodePayloadLenEstimator;
+
+public class MaxValueMetadataNodePayloadLenEstimator implements MetadataNodePayloadLenEstimator {
+
+ private final HashMap<PathType, Integer> maxLenPerPathType = new HashMap<>();
+ private final HashMap<PathType, Integer> maxLenPerListPathType = new HashMap<>();
Review Comment:
> It is not important enough to ensure thread safety for an accurate value, so skipped to guarantee the thread safety
In Java, that's not a proper way to handle thread safety. Another thread might never "see" the changes made by another thread. In practice this rarely happens, but it can happen due to compiler optimizations. That's why such patterns should be avoided in general since they lead to non-deterministic and inconsistent behavior.
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2241532110
##########
pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/MaxValueMetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.zookeeper;
+
+import java.util.HashMap;
+import java.util.List;
+import org.apache.pulsar.common.util.StringSplitter;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataNodePayloadLenEstimator;
+
+public class MaxValueMetadataNodePayloadLenEstimator implements MetadataNodePayloadLenEstimator {
+
+ private final HashMap<PathType, Integer> maxLenPerPathType = new HashMap<>();
+ private final HashMap<PathType, Integer> maxLenPerListPathType = new HashMap<>();
+ // Default to max int value, let the first query command do not execute with batch.
+ private static final int DEFAULT_LEN = 1;//Integer.MAX_VALUE;
+
+ @Override
+ public void recordPut(String path, byte[] data) {
+ PathType pathType = getPathType(path);
+ maxLenPerPathType.compute(pathType, (k, v) -> {
+ if (v == null) {
+ return data.length;
+ }
+ return Math.max(v, data.length);
+ });
+ }
+
+ @Override
+ public void recordGetRes(String path, GetResult getResult) {
+ PathType pathType = getPathType(path);
+ maxLenPerPathType.compute(pathType, (k, v) -> {
+ if (v == null) {
+ return getResult.getValue().length;
+ }
+ return Math.max(v, getResult.getValue().length);
+ });
+ }
+
+ @Override
+ public void recordGetChildrenRes(String path, List<String> list) {
+ PathType pathType = getPathType(path);
+ maxLenPerListPathType.compute(pathType, (k, v) -> {
+ if (v == null) {
+ return list.stream().map(String::length).reduce(0, Integer::sum);
Review Comment:
Addressed
##########
pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/MaxValueMetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.zookeeper;
+
+import java.util.HashMap;
+import java.util.List;
+import org.apache.pulsar.common.util.StringSplitter;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataNodePayloadLenEstimator;
+
+public class MaxValueMetadataNodePayloadLenEstimator implements MetadataNodePayloadLenEstimator {
+
+ private final HashMap<PathType, Integer> maxLenPerPathType = new HashMap<>();
+ private final HashMap<PathType, Integer> maxLenPerListPathType = new HashMap<>();
+ // Default to max int value, let the first query command do not execute with batch.
+ private static final int DEFAULT_LEN = 1;//Integer.MAX_VALUE;
+
+ @Override
+ public void recordPut(String path, byte[] data) {
+ PathType pathType = getPathType(path);
+ maxLenPerPathType.compute(pathType, (k, v) -> {
+ if (v == null) {
+ return data.length;
+ }
+ return Math.max(v, data.length);
+ });
+ }
+
+ @Override
+ public void recordGetRes(String path, GetResult getResult) {
+ PathType pathType = getPathType(path);
+ maxLenPerPathType.compute(pathType, (k, v) -> {
+ if (v == null) {
+ return getResult.getValue().length;
+ }
+ return Math.max(v, getResult.getValue().length);
+ });
+ }
+
+ @Override
+ public void recordGetChildrenRes(String path, List<String> list) {
+ PathType pathType = getPathType(path);
+ maxLenPerListPathType.compute(pathType, (k, v) -> {
+ if (v == null) {
+ return list.stream().map(String::length).reduce(0, Integer::sum);
+ }
+ return Math.max(v, list.stream().map(String::length).reduce(0, Integer::sum));
Review Comment:
Addressed
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#issuecomment-3138851845
> I think that there's another way with ZooKeeper which this fix is targeting.
>
> In current implementation the async version of "multi" is used:
>
> https://github.com/apache/pulsar/blob/4bfdcd85ff5c9f21ba4fd976adf7785465053151/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java#L195-L202
>
> The ZooKeeper client also contains a synchronous version: https://javadoc.io/doc/org.apache.zookeeper/zookeeper/latest/org/apache/zookeeper/ZooKeeper.html#multi(java.lang.Iterable) Using the async multi API causes the problem in the first place. It would be possible to retry the reads by splitting the batch etc until it succeeds.
I checked this and it won't be useful since it would require that pipelining wouldn't be used, which would impact performance a lot. It seems that stats/estimation based approach is the way to go to mitigate the problem.
@poorbarcode I noticed that ZooKeeper has `getNumChildren()` in the stats for ZNodes. Perhaps it would be useful to use that before a namespace listing is performed? That would add latency to the first time the namespace listing is performed, but the number of children could be cached after that. The number of topics in a namespace could vary significantly so that's probably the main source of the problems.
Another source of problems in Pulsar is the ZNode data size for ml cursors (ManagedCursorInfo). I guess that stats/estimation based approach would work very well for that. It would be possible to set the estimated size high so that such operations would be executed in small batches.
Perhaps there could be tunables for configuring the initial estimates for different znode categories?
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2240698420
##########
pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/MaxValueMetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.zookeeper;
+
+import java.util.HashMap;
+import java.util.List;
+import org.apache.pulsar.common.util.StringSplitter;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataNodePayloadLenEstimator;
+
+public class MaxValueMetadataNodePayloadLenEstimator implements MetadataNodePayloadLenEstimator {
+
+ private final HashMap<PathType, Integer> maxLenPerPathType = new HashMap<>();
+ private final HashMap<PathType, Integer> maxLenPerListPathType = new HashMap<>();
+ // Default to max int value, let the first query command do not execute with batch.
+ private static final int DEFAULT_LEN = 1;//Integer.MAX_VALUE;
+
+ @Override
+ public void recordPut(String path, byte[] data) {
+ PathType pathType = getPathType(path);
+ maxLenPerPathType.compute(pathType, (k, v) -> {
+ if (v == null) {
+ return data.length;
+ }
+ return Math.max(v, data.length);
+ });
+ }
+
+ @Override
+ public void recordGetRes(String path, GetResult getResult) {
+ PathType pathType = getPathType(path);
+ maxLenPerPathType.compute(pathType, (k, v) -> {
+ if (v == null) {
+ return getResult.getValue().length;
+ }
+ return Math.max(v, getResult.getValue().length);
+ });
+ }
+
+ @Override
+ public void recordGetChildrenRes(String path, List<String> list) {
+ PathType pathType = getPathType(path);
+ maxLenPerListPathType.compute(pathType, (k, v) -> {
+ if (v == null) {
+ return list.stream().map(String::length).reduce(0, Integer::sum);
+ }
+ return Math.max(v, list.stream().map(String::length).reduce(0, Integer::sum));
+ });
+ }
+
+ @Override
+ public int estimateGetResPayloadLen(String path) {
+ PathType pathType = getPathType(path);
+ return maxLenPerPathType.getOrDefault(pathType, DEFAULT_LEN);
+ }
+
+ @Override
+ public int estimateGetChildrenResPayloadLen(String path) {
+ PathType pathType = getPathType(path);
+ int maxValueCached = maxLenPerListPathType.getOrDefault(pathType, DEFAULT_LEN);
+ return maxValueCached + (maxValueCached >> 3);
Review Comment:
what's the reasoning behind this estimation formula?
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2244185021
##########
pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/MaxValueMetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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.zookeeper;
+
+import io.netty.util.concurrent.FastThreadLocal;
+import java.util.List;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataNodePayloadLenEstimator;
+
+@Slf4j
+public class MaxValueMetadataNodePayloadLenEstimator implements MetadataNodePayloadLenEstimator {
+
+ // Default to max int value, let the first query command do not execute with batch.
+ public static final int DEFAULT_LEN = Integer.MAX_VALUE;
+ public static final int UNSET = -1;
+ // xid: int, zxid: long, err: int, len: int -> 20
+ // Node stat:
+ // czxid: long, mzxid: long, ctime: long, mtime: long, version: int, cversion: int, aversion: int, -> 44
+ // ephemeralOwner: long, dataLength: int, numChildren: int, pzxid: long -> 24
+ // total: 88, let's double it to cover different serialization versions.
+ public static final int ZK_PACKET_SYSTEM_PROPS_LEN = 176;
+ private static final SplitPathRes MEANINGLESS_SPLIT_PATH_RES = new SplitPathRes();
+ private static final FastThreadLocal<SplitPathRes> LOCAL_SPLIT_PATH_RES = new FastThreadLocal<SplitPathRes>() {
+ @Override
+ protected SplitPathRes initialValue() {
+ return new SplitPathRes();
+ }
+ };
+ private final int[] maxLenOfGetMapping;
+ private final int[] maxLenOfListMapping;
+
+ {
+ int pathTypeCount = PathType.values().length;
+ maxLenOfGetMapping = new int[pathTypeCount];
+ maxLenOfListMapping = new int[pathTypeCount];
+ for (int i = 0; i < pathTypeCount; i++) {
+ maxLenOfGetMapping[i] = UNSET;
+ maxLenOfListMapping[i] = UNSET;
+ }
+ }
Review Comment:
Addressed
--
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
Re: [PR] [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24580:
URL: https://github.com/apache/pulsar/pull/24580#discussion_r2247660528
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataNodePayloadLenEstimator.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.metadata.api;
+
+import java.util.List;
+
+public interface MetadataNodePayloadLenEstimator {
+
+ /**
+ * Record the payload length of put operation, to let the estimator know the max payload length, which benefits the
+ * accuracy of estimation.
+ */
+ void recordPut(String path, byte[] data);
+
+ /**
+ * Record the payload length of get result, to let the estimator know the max payload length, which benefits the
+ * accuracy of estimation.
+ */
+ void recordGetRes(String path, GetResult getResult);
+
+ /**
+ * Record the payload length of list result, to let the estimator know the max payload length, which benefits the
+ * accuracy of estimation.
+ */
+ void recordGetChildrenRes(String path, List<String> list);
+
+ /**
+ * Estimate the payload length of get result.
+ */
+ int estimateGetResPayloadLen(String path);
+
+ /**
+ * Estimate the payload length of list result.
+ */
+ int estimateGetChildrenResPayloadLen(String path);
Review Comment:
Good idea, but it leads to the metadata store taking care of a certain use case, which makes the metadata store not clear enough.
--
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