You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by ba...@apache.org on 2023/11/23 10:00:22 UTC
(pulsar) branch branch-2.10 updated: [branch-2.10][fix][broker] Duplicate LedgerOffloader creation when na… (#21610)
This is an automated email from the ASF dual-hosted git repository.
baodi pushed a commit to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.10 by this push:
new f2ff9c59042 [branch-2.10][fix][broker] Duplicate LedgerOffloader creation when na… (#21610)
f2ff9c59042 is described below
commit f2ff9c5904290b78dc49cf6ff98a31dd13ad0319
Author: Baodi Shi <ba...@apache.org>
AuthorDate: Thu Nov 23 18:00:11 2023 +0800
[branch-2.10][fix][broker] Duplicate LedgerOffloader creation when na… (#21610)
---
.../pulsar/broker/admin/TopicAutoCreationTest.java | 2 -
.../common/policies/data/OffloadPoliciesImpl.java | 70 ++++------------------
.../common/policies/data/OffloadPoliciesTest.java | 16 +++++
3 files changed, 26 insertions(+), 62 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAutoCreationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAutoCreationTest.java
index 28020626c4b..3628dfcee18 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAutoCreationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAutoCreationTest.java
@@ -60,8 +60,6 @@ public class TopicAutoCreationTest extends ProducerConsumerBase {
conf.setAllowAutoTopicCreation(true);
conf.setDefaultNumPartitions(3);
conf.setForceDeleteNamespaceAllowed(true);
- conf.setSystemTopicEnabled(true);
- conf.setTopicLevelPoliciesEnabled(true);
super.internalSetup();
super.producerBaseSetup();
}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/OffloadPoliciesImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/OffloadPoliciesImpl.java
index 42f3ccc42e8..0b5288735a1 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/OffloadPoliciesImpl.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/OffloadPoliciesImpl.java
@@ -28,7 +28,6 @@ import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import java.lang.reflect.Field;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
@@ -222,8 +221,7 @@ public class OffloadPoliciesImpl implements Serializable, OffloadPolicies {
public static OffloadPoliciesImpl create(Properties properties) {
OffloadPoliciesImpl data = new OffloadPoliciesImpl();
- Field[] fields = OffloadPoliciesImpl.class.getDeclaredFields();
- Arrays.stream(fields).forEach(f -> {
+ for (Field f : CONFIGURATION_FIELDS) {
if (properties.containsKey(f.getName())) {
try {
f.setAccessible(true);
@@ -234,7 +232,7 @@ public class OffloadPoliciesImpl implements Serializable, OffloadPolicies {
f.getName(), properties.get(f.getName())), e);
}
}
- });
+ }
data.compatibleWithBrokerConfigFile(properties);
return data;
}
@@ -311,62 +309,14 @@ public class OffloadPoliciesImpl implements Serializable, OffloadPolicies {
public Properties toProperties() {
Properties properties = new Properties();
- setProperty(properties, "managedLedgerOffloadedReadPriority", this.getManagedLedgerOffloadedReadPriority());
- setProperty(properties, "offloadersDirectory", this.getOffloadersDirectory());
- setProperty(properties, "managedLedgerOffloadDriver", this.getManagedLedgerOffloadDriver());
- setProperty(properties, "managedLedgerOffloadMaxThreads",
- this.getManagedLedgerOffloadMaxThreads());
- setProperty(properties, "managedLedgerOffloadPrefetchRounds",
- this.getManagedLedgerOffloadPrefetchRounds());
- setProperty(properties, "managedLedgerOffloadThresholdInBytes",
- this.getManagedLedgerOffloadThresholdInBytes());
- setProperty(properties, "managedLedgerOffloadDeletionLagInMillis",
- this.getManagedLedgerOffloadDeletionLagInMillis());
-
- if (this.isS3Driver()) {
- setProperty(properties, "s3ManagedLedgerOffloadRegion",
- this.getS3ManagedLedgerOffloadRegion());
- setProperty(properties, "s3ManagedLedgerOffloadBucket",
- this.getS3ManagedLedgerOffloadBucket());
- setProperty(properties, "s3ManagedLedgerOffloadServiceEndpoint",
- this.getS3ManagedLedgerOffloadServiceEndpoint());
- setProperty(properties, "s3ManagedLedgerOffloadMaxBlockSizeInBytes",
- this.getS3ManagedLedgerOffloadMaxBlockSizeInBytes());
- setProperty(properties, "s3ManagedLedgerOffloadCredentialId",
- this.getS3ManagedLedgerOffloadCredentialId());
- setProperty(properties, "s3ManagedLedgerOffloadCredentialSecret",
- this.getS3ManagedLedgerOffloadCredentialSecret());
- setProperty(properties, "s3ManagedLedgerOffloadRole",
- this.getS3ManagedLedgerOffloadRole());
- setProperty(properties, "s3ManagedLedgerOffloadRoleSessionName",
- this.getS3ManagedLedgerOffloadRoleSessionName());
- setProperty(properties, "s3ManagedLedgerOffloadReadBufferSizeInBytes",
- this.getS3ManagedLedgerOffloadReadBufferSizeInBytes());
- } else if (this.isGcsDriver()) {
- setProperty(properties, "gcsManagedLedgerOffloadRegion",
- this.getGcsManagedLedgerOffloadRegion());
- setProperty(properties, "gcsManagedLedgerOffloadBucket",
- this.getGcsManagedLedgerOffloadBucket());
- setProperty(properties, "gcsManagedLedgerOffloadMaxBlockSizeInBytes",
- this.getGcsManagedLedgerOffloadMaxBlockSizeInBytes());
- setProperty(properties, "gcsManagedLedgerOffloadReadBufferSizeInBytes",
- this.getGcsManagedLedgerOffloadReadBufferSizeInBytes());
- setProperty(properties, "gcsManagedLedgerOffloadServiceAccountKeyFile",
- this.getGcsManagedLedgerOffloadServiceAccountKeyFile());
- } else if (this.isFileSystemDriver()) {
- setProperty(properties, "fileSystemProfilePath", this.getFileSystemProfilePath());
- setProperty(properties, "fileSystemURI", this.getFileSystemURI());
- }
-
- setProperty(properties, "managedLedgerOffloadBucket", this.getManagedLedgerOffloadBucket());
- setProperty(properties, "managedLedgerOffloadRegion", this.getManagedLedgerOffloadRegion());
- setProperty(properties, "managedLedgerOffloadServiceEndpoint",
- this.getManagedLedgerOffloadServiceEndpoint());
- setProperty(properties, "managedLedgerOffloadMaxBlockSizeInBytes",
- this.getManagedLedgerOffloadMaxBlockSizeInBytes());
- setProperty(properties, "managedLedgerOffloadReadBufferSizeInBytes",
- this.getManagedLedgerOffloadReadBufferSizeInBytes());
-
+ for (Field f : CONFIGURATION_FIELDS) {
+ try {
+ f.setAccessible(true);
+ setProperty(properties, f.getName(), f.get(this));
+ } catch (Exception e) {
+ throw new IllegalArgumentException("An error occurred while processing the field: " + f.getName(), e);
+ }
+ }
return properties;
}
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/OffloadPoliciesTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/OffloadPoliciesTest.java
index f9c856e69d5..0375abb207d 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/OffloadPoliciesTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/OffloadPoliciesTest.java
@@ -305,4 +305,20 @@ public class OffloadPoliciesTest {
Assert.assertNull(offloadPolicies.getS3ManagedLedgerOffloadRegion());
}
+ /**
+ * Test toProperties as well as create from properties.
+ * @throws Exception
+ */
+ @Test
+ public void testToProperties() throws Exception {
+ // Base information convert.
+ OffloadPoliciesImpl offloadPolicies = OffloadPoliciesImpl.create("aws-s3", "test-region", "test-bucket",
+ "http://test.endpoint",null, null, null, null, 32 * 1024 * 1024, 5 * 1024 * 1024,
+ 10 * 1024 * 1024L, 10000L, OffloadedReadPriority.TIERED_STORAGE_FIRST);
+ Assert.assertEquals(offloadPolicies, OffloadPoliciesImpl.create(offloadPolicies.toProperties()));
+
+ // Set useless config to offload policies. Make sure convert conversion result is the same.
+ offloadPolicies.setFileSystemProfilePath("/test/file");
+ Assert.assertEquals(offloadPolicies, OffloadPoliciesImpl.create(offloadPolicies.toProperties()));
+ }
}