You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/02/20 08:50:50 UTC
(pulsar) branch branch-3.3 updated (2fb10d8e0ba -> a622a6a5d63)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 2fb10d8e0ba Release 3.3.5
new a1cfbc3a996 [improve][broker] Fix non-persistent system topic schema compatibility (#23286)
new a622a6a5d63 [improve][test] Upgrade Testcontainers to 1.20.4 and docker-java to 3.4.0 (#24003)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
pom.xml | 7 ++--
.../pulsar/broker/service/BrokerService.java | 7 +++-
...age-info.java => NonPersistentSystemTopic.java} | 13 +++++++
.../extensions/ExtensibleLoadManagerImplTest.java | 42 ++++++++++++++++++++++
4 files changed, 64 insertions(+), 5 deletions(-)
copy pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/{package-info.java => NonPersistentSystemTopic.java} (72%)
(pulsar) 01/02: [improve][broker] Fix non-persistent system topic schema compatibility (#23286)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a1cfbc3a996afab23bfaad8d391b1f8049f6f29d
Author: Kai Wang <kw...@apache.org>
AuthorDate: Wed Feb 19 21:41:07 2025 +0800
[improve][broker] Fix non-persistent system topic schema compatibility (#23286)
When upgrading broker version from `3.0.x` to `3.3.x` with `ExtensibleLoadManagerImpl` enabled, it will have an `Unable to read schema` exception. And the broker will fail to start. This issue is caused by https://github.com/apache/pulsar/pull/22055 .
Add a new class `NonPersistentSystemTopic`, and it will use for system non-persistent topic.
(cherry picked from commit 7dbd8a507fc07d4fd5dc813ad76ee8080afd43cb)
---
.../pulsar/broker/service/BrokerService.java | 7 +++-
.../nonpersistent/NonPersistentSystemTopic.java | 32 +++++++++++++++++
.../extensions/ExtensibleLoadManagerImplTest.java | 42 ++++++++++++++++++++++
3 files changed, 80 insertions(+), 1 deletion(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index c123673dd32..1e476f8e77d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -121,6 +121,7 @@ import org.apache.pulsar.broker.service.BrokerServiceException.ServiceUnitNotRea
import org.apache.pulsar.broker.service.BrokerServiceException.TopicMigratedException;
import org.apache.pulsar.broker.service.TopicEventsListener.EventStage;
import org.apache.pulsar.broker.service.TopicEventsListener.TopicEvent;
+import org.apache.pulsar.broker.service.nonpersistent.NonPersistentSystemTopic;
import org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopic;
import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter;
import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
@@ -1286,7 +1287,11 @@ public class BrokerService implements Closeable {
final long topicCreateTimeMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime());
NonPersistentTopic nonPersistentTopic;
try {
- nonPersistentTopic = newTopic(topic, null, this, NonPersistentTopic.class);
+ if (isSystemTopic(topic)) {
+ nonPersistentTopic = new NonPersistentSystemTopic(topic, this);
+ } else {
+ nonPersistentTopic = newTopic(topic, null, this, NonPersistentTopic.class);
+ }
nonPersistentTopic.setCreateFuture(topicFuture);
} catch (Throwable e) {
log.warn("Failed to create topic {}", topic, e);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSystemTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSystemTopic.java
new file mode 100644
index 00000000000..9b867c9a8b3
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSystemTopic.java
@@ -0,0 +1,32 @@
+/*
+ * 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.nonpersistent;
+
+import org.apache.pulsar.broker.service.BrokerService;
+
+public class NonPersistentSystemTopic extends NonPersistentTopic {
+ public NonPersistentSystemTopic(String topic, BrokerService brokerService) {
+ super(topic, brokerService);
+ }
+
+ @Override
+ public boolean isSystemTopic() {
+ return true;
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
index ecba040d4f6..b394278eb19 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
@@ -81,6 +81,7 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import lombok.Cleanup;
+import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.mutable.MutableInt;
import org.apache.commons.lang3.reflect.FieldUtils;
@@ -103,6 +104,7 @@ import org.apache.pulsar.broker.loadbalance.extensions.models.SplitDecision;
import org.apache.pulsar.broker.loadbalance.extensions.models.UnloadCounter;
import org.apache.pulsar.broker.loadbalance.extensions.reporter.BrokerLoadDataReporter;
import org.apache.pulsar.broker.loadbalance.extensions.scheduler.TransferShedder;
+import org.apache.pulsar.broker.loadbalance.extensions.store.LoadDataStoreFactory;
import org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl;
import org.apache.pulsar.broker.lookup.LookupResult;
import org.apache.pulsar.broker.namespace.LookupOptions;
@@ -111,6 +113,7 @@ import org.apache.pulsar.broker.namespace.NamespaceBundleSplitListener;
import org.apache.pulsar.broker.namespace.NamespaceEphemeralData;
import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.broker.service.BrokerServiceException;
+import org.apache.pulsar.broker.service.nonpersistent.NonPersistentSystemTopic;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.Consumer;
@@ -131,6 +134,7 @@ import org.apache.pulsar.common.naming.TopicVersion;
import org.apache.pulsar.common.policies.data.BrokerAssignment;
import org.apache.pulsar.common.policies.data.BundlesData;
import org.apache.pulsar.common.policies.data.NamespaceOwnershipStatus;
+import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
import org.apache.pulsar.common.stats.Metrics;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.metadata.api.MetadataStoreException;
@@ -138,6 +142,7 @@ import org.apache.pulsar.policies.data.loadbalancer.NamespaceBundleStats;
import org.apache.pulsar.policies.data.loadbalancer.ResourceUsage;
import org.apache.pulsar.policies.data.loadbalancer.SystemResourceUsage;
import org.awaitility.Awaitility;
+import org.testng.Assert;
import org.testng.AssertJUnit;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@@ -1924,6 +1929,43 @@ public class ExtensibleLoadManagerImplTest extends ExtensibleLoadManagerImplBase
});
}
+ @Test
+ public void testSystemNonPersistentTopicSchemaCompatibility() throws Exception {
+ String topicName = ExtensibleLoadManagerImpl.BROKER_LOAD_DATA_STORE_TOPIC;
+ NonPersistentSystemTopic topic = new NonPersistentSystemTopic(topicName, pulsar.getBrokerService());
+ Assert.assertEquals(SchemaCompatibilityStrategy.ALWAYS_COMPATIBLE, topic.getSchemaCompatibilityStrategy());
+
+ var brokerLoadDataStore = LoadDataStoreFactory.create(pulsar, topicName, BrokerLoadDataV1.class);
+ brokerLoadDataStore.init();
+ brokerLoadDataStore.pushAsync("key", new BrokerLoadDataV1()).get();
+ Awaitility.await().until(() -> {
+ var data = brokerLoadDataStore.get("key");
+ return data.isPresent();
+ });
+ brokerLoadDataStore.pushAsync("key", null).get();
+ brokerLoadDataStore.close();
+ }
+
+ @Data
+ private static class BrokerLoadDataV1 {
+ private ResourceUsage cpu;
+ private ResourceUsage memory;
+ private ResourceUsage directMemory;
+ private ResourceUsage bandwidthIn;
+ private ResourceUsage bandwidthOut;
+ private double msgThroughputIn;
+ private double msgThroughputOut;
+ private double msgRateIn;
+ private double msgRateOut;
+ private int bundleCount;
+ private int topics;
+ private double maxResourceUsage;
+ private double weightedMaxEMA;
+ private double msgThroughputEMA;
+ private long updatedAt;
+ private long reportedAt;
+ }
+
@Test(timeOut = 10 * 1000, priority = 5000)
public void unloadTimeoutCheckTest()
throws Exception {
(pulsar) 02/02: [improve][test] Upgrade Testcontainers to 1.20.4 and docker-java to 3.4.0 (#24003)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a622a6a5d63cc59f122b84c143051c19ae33d640
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Feb 20 03:34:55 2025 +0200
[improve][test] Upgrade Testcontainers to 1.20.4 and docker-java to 3.4.0 (#24003)
(cherry picked from commit 1c6aa7cdb04f1ad9625a3292897e3236a574cc86)
---
pom.xml | 7 +++----
1 file changed, 3 insertions(+), 4 deletions(-)
diff --git a/pom.xml b/pom.xml
index 46ab340630f..3e1cabd8dfe 100644
--- a/pom.xml
+++ b/pom.xml
@@ -264,12 +264,11 @@ flexible messaging model and an intuitive client API.</description>
<failsafe.version>3.3.2</failsafe.version>
<!-- test dependencies -->
- <testcontainers.version>1.18.3</testcontainers.version>
+ <testcontainers.version>1.20.4</testcontainers.version>
+ <!-- Set docker-java.version to the version of docker-java used in org.testcontainers:testcontainers pom -->
+ <docker-java.version>3.4.0</docker-java.version>
<hamcrest.version>2.2</hamcrest.version>
<restassured.version>5.4.0</restassured.version>
-
- <!-- Set docker-java.version to the version of docker-java used in Testcontainers -->
- <docker-java.version>3.3.0</docker-java.version>
<kerby.version>1.1.1</kerby.version>
<testng.version>7.7.1</testng.version>
<mockito.version>5.6.0</mockito.version>