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 2024/11/18 08:00:05 UTC
(pulsar) branch branch-4.0 updated (3f949a79500 -> a6fabbb456a)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 3f949a79500 [improve][test] Clarify method signatures in Bookkeeper mock client (#23598)
new 8f93f545bc8 [fix][misc] Unable to connect an etcd metastore with recent releases due to jetc-core sharding problem (#23604)
new 6e5aab9e03a [improve][offload] Use filesystemURI as the storage path (#23591)
new a6fabbb456a [fix][broker] Fix failed TokenAuthenticatedProducerConsumerTest (#23602)
The 3 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:
distribution/server/pom.xml | 1 -
jetcd-core-shaded/pom.xml | 49 +--------
pom.xml | 1 -
pulsar-broker/pom.xml | 1 -
.../TokenAuthenticatedProducerConsumerTest.java | 4 +-
pulsar-metadata/pom.xml | 1 -
.../impl/FileSystemManagedLedgerOffloader.java | 2 +-
.../impl/FileSystemOffloaderLocalFileTest.java | 119 +++++++++++++++++++++
.../resources}/filesystem_offload_core_site.xml | 0
9 files changed, 123 insertions(+), 55 deletions(-)
create mode 100644 tiered-storage/file-system/src/test/java/org/apache/bookkeeper/mledger/offload/filesystem/impl/FileSystemOffloaderLocalFileTest.java
copy {conf => tiered-storage/file-system/src/test/resources}/filesystem_offload_core_site.xml (100%)
(pulsar) 02/03: [improve][offload] Use filesystemURI as the storage path (#23591)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 6e5aab9e03a47a910197ba80b5e089f6055824f0
Author: Yong Zhang <zh...@gmail.com>
AuthorDate: Mon Nov 18 10:09:26 2024 +0800
[improve][offload] Use filesystemURI as the storage path (#23591)
(cherry picked from commit b915f6e73e5b8cfa94c4c55eb930befc682674d4)
---
.../impl/FileSystemManagedLedgerOffloader.java | 2 +-
.../impl/FileSystemOffloaderLocalFileTest.java | 119 +++++++++++++++++++++
.../resources/filesystem_offload_core_site.xml | 48 +++++++++
3 files changed, 168 insertions(+), 1 deletion(-)
diff --git a/tiered-storage/file-system/src/main/java/org/apache/bookkeeper/mledger/offload/filesystem/impl/FileSystemManagedLedgerOffloader.java b/tiered-storage/file-system/src/main/java/org/apache/bookkeeper/mledger/offload/filesystem/impl/FileSystemManagedLedgerOffloader.java
index 56612adc1ef..2431e61c91c 100644
--- a/tiered-storage/file-system/src/main/java/org/apache/bookkeeper/mledger/offload/filesystem/impl/FileSystemManagedLedgerOffloader.java
+++ b/tiered-storage/file-system/src/main/java/org/apache/bookkeeper/mledger/offload/filesystem/impl/FileSystemManagedLedgerOffloader.java
@@ -107,7 +107,7 @@ public class FileSystemManagedLedgerOffloader implements LedgerOffloader {
this.configuration.setClassLoader(FileSystemLedgerOffloaderFactory.class.getClassLoader());
this.driverName = conf.getManagedLedgerOffloadDriver();
- this.storageBasePath = configuration.get("hadoop.tmp.dir");
+ this.storageBasePath = configuration.get("fs.defaultFS");
this.scheduler = scheduler;
this.fileSystem = FileSystem.get(configuration);
this.assignmentScheduler = OrderedScheduler.newSchedulerBuilder()
diff --git a/tiered-storage/file-system/src/test/java/org/apache/bookkeeper/mledger/offload/filesystem/impl/FileSystemOffloaderLocalFileTest.java b/tiered-storage/file-system/src/test/java/org/apache/bookkeeper/mledger/offload/filesystem/impl/FileSystemOffloaderLocalFileTest.java
new file mode 100644
index 00000000000..14734b3faca
--- /dev/null
+++ b/tiered-storage/file-system/src/test/java/org/apache/bookkeeper/mledger/offload/filesystem/impl/FileSystemOffloaderLocalFileTest.java
@@ -0,0 +1,119 @@
+/*
+ * 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.bookkeeper.mledger.offload.filesystem.impl;
+
+import static org.testng.Assert.assertEquals;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.UUID;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.client.PulsarMockBookKeeper;
+import org.apache.bookkeeper.client.api.DigestType;
+import org.apache.bookkeeper.client.api.LedgerEntries;
+import org.apache.bookkeeper.client.api.LedgerEntry;
+import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.bookkeeper.common.util.OrderedScheduler;
+import org.apache.bookkeeper.mledger.LedgerOffloaderStats;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.OffloadPoliciesImpl;
+import org.testng.annotations.Test;
+
+public class FileSystemOffloaderLocalFileTest {
+ private OrderedScheduler scheduler = OrderedScheduler.newSchedulerBuilder().numThreads(1).name("offloader").build();
+ private LedgerOffloaderStats offloaderStats = LedgerOffloaderStats.create(true, true, scheduler, 60);
+
+
+ private String getResourceFilePath(String name) {
+ return getClass().getClassLoader().getResource(name).getPath();
+ }
+
+ @Test
+ public void testReadWriteWithLocalFileUsingFileSystemURI() throws Exception {
+ // prepare the offload policies
+ final String basePath = "/tmp";
+ OffloadPoliciesImpl offloadPolicies = new OffloadPoliciesImpl();
+ offloadPolicies.setFileSystemURI("file://" + basePath);
+ offloadPolicies.setManagedLedgerOffloadDriver("filesystem");
+ offloadPolicies.setFileSystemProfilePath(getResourceFilePath("filesystem_offload_core_site.xml"));
+
+ // initialize the offloader with the offload policies
+ var offloader = FileSystemManagedLedgerOffloader.create(offloadPolicies, scheduler, offloaderStats);
+
+ int numberOfEntries = 100;
+
+ // prepare the data in bookkeeper
+ BookKeeper bk = new PulsarMockBookKeeper(scheduler);
+ LedgerHandle lh = bk.createLedger(1,1,1, BookKeeper.DigestType.CRC32, "".getBytes());
+ for (int i = 0; i < numberOfEntries; i++) {
+ byte[] entry = ("foobar"+i).getBytes();
+ lh.addEntry(entry);
+ }
+ lh.close();
+
+ ReadHandle read = bk.newOpenLedgerOp()
+ .withLedgerId(lh.getId())
+ .withDigestType(DigestType.CRC32)
+ .withPassword("".getBytes()).execute().get();
+
+ final String mlName = TopicName.get("testWriteLocalFIle").getPersistenceNamingEncoding();
+ Map<String, String> offloadDriverMetadata = new HashMap<>();
+ offloadDriverMetadata.put("ManagedLedgerName", mlName);
+
+ UUID uuid = UUID.randomUUID();
+ offloader.offload(read, uuid, offloadDriverMetadata).get();
+ ReadHandle toTest = offloader.readOffloaded(read.getId(), uuid, offloadDriverMetadata).get();
+ assertEquals(toTest.getLastAddConfirmed(), read.getLastAddConfirmed());
+ LedgerEntries toTestEntries = toTest.read(0, numberOfEntries - 1);
+ LedgerEntries toWriteEntries = read.read(0,numberOfEntries - 1);
+ Iterator<LedgerEntry> toTestIter = toTestEntries.iterator();
+ Iterator<LedgerEntry> toWriteIter = toWriteEntries.iterator();
+ while(toTestIter.hasNext()) {
+ LedgerEntry toWriteEntry = toWriteIter.next();
+ LedgerEntry toTestEntry = toTestIter.next();
+
+ assertEquals(toWriteEntry.getLedgerId(), toTestEntry.getLedgerId());
+ assertEquals(toWriteEntry.getEntryId(), toTestEntry.getEntryId());
+ assertEquals(toWriteEntry.getLength(), toTestEntry.getLength());
+ assertEquals(toWriteEntry.getEntryBuffer(), toTestEntry.getEntryBuffer());
+ }
+ toTestEntries = toTest.read(1, numberOfEntries - 1);
+ toWriteEntries = read.read(1,numberOfEntries - 1);
+ toTestIter = toTestEntries.iterator();
+ toWriteIter = toWriteEntries.iterator();
+ while(toTestIter.hasNext()) {
+ LedgerEntry toWriteEntry = toWriteIter.next();
+ LedgerEntry toTestEntry = toTestIter.next();
+
+ assertEquals(toWriteEntry.getLedgerId(), toTestEntry.getLedgerId());
+ assertEquals(toWriteEntry.getEntryId(), toTestEntry.getEntryId());
+ assertEquals(toWriteEntry.getLength(), toTestEntry.getLength());
+ assertEquals(toWriteEntry.getEntryBuffer(), toTestEntry.getEntryBuffer());
+ }
+
+ // check the file located in the local file system
+ Path offloadedFilePath = Paths.get(basePath, mlName);
+ assertEquals(Files.exists(offloadedFilePath), true);
+ }
+}
diff --git a/tiered-storage/file-system/src/test/resources/filesystem_offload_core_site.xml b/tiered-storage/file-system/src/test/resources/filesystem_offload_core_site.xml
new file mode 100644
index 00000000000..d26cec2cc60
--- /dev/null
+++ b/tiered-storage/file-system/src/test/resources/filesystem_offload_core_site.xml
@@ -0,0 +1,48 @@
+<!--
+
+ 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.
+
+-->
+<configuration>
+ <!--file system uri, necessary-->
+ <property>
+ <name>fs.defaultFS</name>
+ <value></value>
+ </property>
+ <property>
+ <name>hadoop.tmp.dir</name>
+ <value>pulsar</value>
+ </property>
+ <property>
+ <name>io.file.buffer.size</name>
+ <value>4096</value>
+ </property>
+ <property>
+ <name>io.seqfile.compress.blocksize</name>
+ <value>1000000</value>
+ </property>
+ <property>
+ <name>io.seqfile.compression.type</name>
+ <value>BLOCK</value>
+ </property>
+ <property>
+ <name>io.map.index.interval</name>
+ <value>128</value>
+ </property>
+
+</configuration>
(pulsar) 03/03: [fix][broker] Fix failed TokenAuthenticatedProducerConsumerTest (#23602)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a6fabbb456a2eca5689511996ff5e5abbd92e3a5
Author: Yunze Xu <xy...@163.com>
AuthorDate: Mon Nov 18 10:18:30 2024 +0800
[fix][broker] Fix failed TokenAuthenticatedProducerConsumerTest (#23602)
(cherry picked from commit 27158532de26d40e7a402769e73ddd3be43f0623)
---
.../pulsar/client/api/TokenAuthenticatedProducerConsumerTest.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TokenAuthenticatedProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TokenAuthenticatedProducerConsumerTest.java
index 9da3fcbd0ed..a9a568f1326 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TokenAuthenticatedProducerConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TokenAuthenticatedProducerConsumerTest.java
@@ -191,14 +191,14 @@ public class TokenAuthenticatedProducerConsumerTest extends ProducerConsumerBase
}
@Test(dataProvider = "provider")
- public void testTopicNotFound(boolean useTcpServiceUrl, boolean useCorrectToken) throws Exception {
+ public void testTenantNotExist(boolean useTcpServiceUrl, boolean useCorrectToken) throws Exception {
final var operationTimeoutMs = 10000;
final var url = useTcpServiceUrl ? pulsar.getBrokerServiceUrl() : pulsar.getWebServiceAddress();
final var token = useCorrectToken ? ADMIN_TOKEN : USER_TOKEN;
@Cleanup final var client = PulsarClient.builder().serviceUrl(url)
.operationTimeout(operationTimeoutMs, TimeUnit.MILLISECONDS)
.authentication(AuthenticationFactory.token(token)).build();
- final var topic = "my-property/not-exist/tp"; // the namespace does not exist
+ final var topic = "non-exist/not-exist/tp"; // the namespace does not exist
var start = System.currentTimeMillis();
try {
client.newProducer().topic(topic).create();
(pulsar) 01/03: [fix][misc] Unable to connect an etcd metastore with recent releases due to jetc-core sharding problem (#23604)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 8f93f545bc8287daaeab5359f4f9b4c3601c1e8f
Author: 萧易客 <km...@live.com>
AuthorDate: Sun Nov 17 02:45:12 2024 +0800
[fix][misc] Unable to connect an etcd metastore with recent releases due to jetc-core sharding problem (#23604)
Co-authored-by: Lari Hotari <lh...@users.noreply.github.com>
(cherry picked from commit 89ccb7361b80dfb7897ea2555dfb6e884f62b916)
---
distribution/server/pom.xml | 1 -
jetcd-core-shaded/pom.xml | 49 +--------------------------------------------
pom.xml | 1 -
pulsar-broker/pom.xml | 1 -
pulsar-metadata/pom.xml | 1 -
5 files changed, 1 insertion(+), 52 deletions(-)
diff --git a/distribution/server/pom.xml b/distribution/server/pom.xml
index fe799031b97..bd1538e5bf6 100644
--- a/distribution/server/pom.xml
+++ b/distribution/server/pom.xml
@@ -49,7 +49,6 @@
<groupId>${project.groupId}</groupId>
<artifactId>jetcd-core-shaded</artifactId>
<version>${project.version}</version>
- <classifier>shaded</classifier>
</dependency>
<dependency>
diff --git a/jetcd-core-shaded/pom.xml b/jetcd-core-shaded/pom.xml
index 2d9263472ea..92cf8e7782c 100644
--- a/jetcd-core-shaded/pom.xml
+++ b/jetcd-core-shaded/pom.xml
@@ -74,6 +74,7 @@
</dependency>
</dependencies>
<build>
+ <finalName>${project.artifactId}-${project.version}</finalName>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
@@ -141,54 +142,6 @@
<file>${project.basedir}/dependency-reduced-pom.xml</file>
</transformer>
</transformers>
- <!-- required for IntelliJ support -->
- <shadedArtifactAttached>true</shadedArtifactAttached>
- <shadedClassifierName>shaded</shadedClassifierName>
- </configuration>
- </execution>
- </executions>
- </plugin>
- <!-- required for IntelliJ support, for some reason shadedArtifactAttached isn't sufficient alone -->
- <plugin>
- <groupId>org.codehaus.mojo</groupId>
- <artifactId>build-helper-maven-plugin</artifactId>
- <executions>
- <execution>
- <id>attach-shaded-jar</id>
- <phase>package</phase>
- <goals>
- <goal>attach-artifact</goal>
- </goals>
- <configuration>
- <artifacts>
- <artifact>
- <file>${project.build.directory}/${project.artifactId}-${project.version}-shaded.jar</file>
- <type>jar</type>
- <classifier>shaded</classifier>
- </artifact>
- </artifacts>
- </configuration>
- </execution>
- </executions>
- </plugin>
- <!-- required for running tests in subproject -->
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-antrun-plugin</artifactId>
- <version>${maven-antrun-plugin.version}</version>
- <executions>
- <execution>
- <id>unpack-shaded-jar</id>
- <phase>package</phase>
- <goals>
- <goal>run</goal>
- </goals>
- <configuration>
- <target>
- <unzip src="${project.build.directory}/${project.artifactId}-${project.version}-shaded.jar"
- dest="${project.build.outputDirectory}"
- overwrite="true" />
- </target>
</configuration>
</execution>
</executions>
diff --git a/pom.xml b/pom.xml
index df415b3c7b6..5a939453096 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1108,7 +1108,6 @@ flexible messaging model and an intuitive client API.</description>
<groupId>${project.groupId}</groupId>
<artifactId>jetcd-core-shaded</artifactId>
<version>${project.version}</version>
- <classifier>shaded</classifier>
<exclusions>
<exclusion>
<groupId>io.etcd</groupId>
diff --git a/pulsar-broker/pom.xml b/pulsar-broker/pom.xml
index 7e64623c831..8f00407af3c 100644
--- a/pulsar-broker/pom.xml
+++ b/pulsar-broker/pom.xml
@@ -487,7 +487,6 @@
<groupId>${project.groupId}</groupId>
<artifactId>jetcd-core-shaded</artifactId>
<version>${project.version}</version>
- <classifier>shaded</classifier>
<scope>test</scope>
</dependency>
<dependency>
diff --git a/pulsar-metadata/pom.xml b/pulsar-metadata/pom.xml
index dd3b4dd3160..f422f21fc5e 100644
--- a/pulsar-metadata/pom.xml
+++ b/pulsar-metadata/pom.xml
@@ -125,7 +125,6 @@
<groupId>${project.groupId}</groupId>
<artifactId>jetcd-core-shaded</artifactId>
<version>${project.version}</version>
- <classifier>shaded</classifier>
</dependency>
<dependency>
<groupId>io.grpc</groupId>