From ebe32aba90a8f7c3671a20d7be8a71d4d849a83b Mon Sep 17 00:00:00 2001 From: Baodi Shi Date: Mon, 20 Nov 2023 17:17:28 +0800 Subject: [PATCH] =?UTF-8?q?[fix][broker]=20Duplicate=20LedgerOffloader=20c?= =?UTF-8?q?reation=20when=20namespace/topic=E2=80=A6=20(#21591)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit (cherry picked from commit 98bf9dd72910e1b02dea17148a4199e3b26d7147) --- .../policies/data/OffloadPoliciesImpl.java | 72 ++++--------------- .../policies/data/OffloadPoliciesTest.java | 17 +++++ 2 files changed, 29 insertions(+), 60 deletions(-) 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 42f3ccc42e8ac1..8c09f79a05ae22 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 @@ -30,6 +30,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.HashMap; import java.util.List; import java.util.Properties; import lombok.Data; @@ -70,6 +71,7 @@ public class OffloadPoliciesImpl implements Serializable, OffloadPolicies { public static final String DEFAULT_OFFLOADER_DIRECTORY = "./offloaders"; public static final Long DEFAULT_OFFLOAD_THRESHOLD_IN_BYTES = null; public static final Long DEFAULT_OFFLOAD_DELETION_LAG_IN_MILLIS = null; + public static final String EXTRA_CONFIG_PREFIX = "managedLedgerOffloadExtraConfig"; public static final String OFFLOAD_THRESHOLD_NAME_IN_CONF_FILE = "managedLedgerOffloadAutoTriggerSizeThresholdBytes"; @@ -99,7 +101,6 @@ public class OffloadPoliciesImpl implements Serializable, OffloadPolicies { @Configuration @JsonProperty(access = JsonProperty.Access.READ_WRITE) private OffloadedReadPriority managedLedgerOffloadedReadPriority = DEFAULT_OFFLOADED_READ_PRIORITY; - // s3 config, set by service configuration or cli @Configuration @JsonProperty(access = JsonProperty.Access.READ_WRITE) @@ -222,8 +223,7 @@ public static OffloadPoliciesImpl create(String driver, String region, String bu 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 +234,7 @@ public static OffloadPoliciesImpl create(Properties properties) { f.getName(), properties.get(f.getName())), e); } } - }); + } data.compatibleWithBrokerConfigFile(properties); return data; } @@ -311,62 +311,14 @@ public boolean bucketValid() { 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 f9c856e69d56b7..ce5dac67f06358 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 @@ -18,6 +18,7 @@ */ package org.apache.pulsar.common.policies.data; +import static org.testng.Assert.assertEquals; import java.util.Properties; import org.testng.Assert; import org.testng.annotations.Test; @@ -305,4 +306,20 @@ public void mergeTest() { 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, 100L, 10000L, OffloadedReadPriority.TIERED_STORAGE_FIRST); + assertEquals(offloadPolicies, OffloadPoliciesImpl.create(offloadPolicies.toProperties())); + + // Set useless config to offload policies. Make sure convert conversion result is the same. + offloadPolicies.setFileSystemProfilePath("/test/file"); + assertEquals(offloadPolicies, OffloadPoliciesImpl.create(offloadPolicies.toProperties())); + } }