Skip to content

Commit c5fd8f0

Browse files
zymapzymap
committed
[improve][offload] Use filesystemURI as the storage path
--- ### Motivation We provided the fileSystemUri in the offload policy as the filesystem offload configuration. The fileSystemUri will overwrite the fs.defaultFS. We should use it as the storage path not the hadoop.tmp.dir as the storage path.
1 parent 7822dca commit c5fd8f0

File tree

3 files changed

+168
-1
lines changed

3 files changed

+168
-1
lines changed

‎tiered-storage/file-system/src/main/java/org/apache/bookkeeper/mledger/offload/filesystem/impl/FileSystemManagedLedgerOffloader.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -107,7 +107,7 @@ private FileSystemManagedLedgerOffloader(OffloadPoliciesImpl conf, OrderedSchedu
107107

108108
this.configuration.setClassLoader(FileSystemLedgerOffloaderFactory.class.getClassLoader());
109109
this.driverName = conf.getManagedLedgerOffloadDriver();
110-
this.storageBasePath = configuration.get("hadoop.tmp.dir");
110+
this.storageBasePath = configuration.get("fs.defaultFS");
111111
this.scheduler = scheduler;
112112
this.fileSystem = FileSystem.get(configuration);
113113
this.assignmentScheduler = OrderedScheduler.newSchedulerBuilder()

‎tiered-storage/file-system/src/test/java/org/apache/bookkeeper/mledger/offload/filesystem/impl/FileSystemOffloaderLocalFileTest.java

+119
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,119 @@
1+
/**
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
* <p>
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
* <p>
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
package org.apache.bookkeeper.mledger.offload.filesystem.impl;
20+
21+
import static org.testng.Assert.assertEquals;
22+
23+
import java.nio.file.Files;
24+
import java.nio.file.Path;
25+
import java.nio.file.Paths;
26+
import java.util.HashMap;
27+
import java.util.Iterator;
28+
import java.util.Map;
29+
import java.util.UUID;
30+
import org.apache.bookkeeper.client.BookKeeper;
31+
import org.apache.bookkeeper.client.LedgerHandle;
32+
import org.apache.bookkeeper.client.PulsarMockBookKeeper;
33+
import org.apache.bookkeeper.client.api.DigestType;
34+
import org.apache.bookkeeper.client.api.LedgerEntries;
35+
import org.apache.bookkeeper.client.api.LedgerEntry;
36+
import org.apache.bookkeeper.client.api.ReadHandle;
37+
import org.apache.bookkeeper.common.util.OrderedScheduler;
38+
import org.apache.bookkeeper.mledger.LedgerOffloaderStats;
39+
import org.apache.pulsar.common.naming.TopicName;
40+
import org.apache.pulsar.common.policies.data.OffloadPoliciesImpl;
41+
import org.testng.annotations.Test;
42+
43+
public class FileSystemOffloaderLocalFileTest {
44+
private OrderedScheduler scheduler = OrderedScheduler.newSchedulerBuilder().numThreads(1).name("offloader").build();
45+
private LedgerOffloaderStats offloaderStats = LedgerOffloaderStats.create(true, true, scheduler, 60);
46+
47+
48+
private String getResourceFilePath(String name) {
49+
return getClass().getClassLoader().getResource(name).getPath();
50+
}
51+
52+
@Test
53+
public void testReadWriteWithLocalFileUsingFileSystemURI() throws Exception {
54+
// prepare the offload policies
55+
final String basePath = "/tmp";
56+
OffloadPoliciesImpl offloadPolicies = new OffloadPoliciesImpl();
57+
offloadPolicies.setFileSystemURI("file://" + basePath);
58+
offloadPolicies.setManagedLedgerOffloadDriver("filesystem");
59+
offloadPolicies.setFileSystemProfilePath(getResourceFilePath("filesystem_offload_core_site.xml"));
60+
61+
// initialize the offloader with the offload policies
62+
var offloader = FileSystemManagedLedgerOffloader.create(offloadPolicies, scheduler, offloaderStats);
63+
64+
int numberOfEntries = 100;
65+
66+
// prepare the data in bookkeeper
67+
BookKeeper bk = new PulsarMockBookKeeper(scheduler);
68+
LedgerHandle lh = bk.createLedger(1,1,1, BookKeeper.DigestType.CRC32, "".getBytes());
69+
for (int i = 0; i < numberOfEntries; i++) {
70+
byte[] entry = ("foobar"+i).getBytes();
71+
lh.addEntry(entry);
72+
}
73+
lh.close();
74+
75+
ReadHandle read = bk.newOpenLedgerOp()
76+
.withLedgerId(lh.getId())
77+
.withDigestType(DigestType.CRC32)
78+
.withPassword("".getBytes()).execute().get();
79+
80+
final String mlName = TopicName.get("testWriteLocalFIle").getPersistenceNamingEncoding();
81+
Map<String, String> offloadDriverMetadata = new HashMap<>();
82+
offloadDriverMetadata.put("ManagedLedgerName", mlName);
83+
84+
UUID uuid = UUID.randomUUID();
85+
offloader.offload(read, uuid, offloadDriverMetadata).get();
86+
ReadHandle toTest = offloader.readOffloaded(read.getId(), uuid, offloadDriverMetadata).get();
87+
assertEquals(toTest.getLastAddConfirmed(), read.getLastAddConfirmed());
88+
LedgerEntries toTestEntries = toTest.read(0, numberOfEntries - 1);
89+
LedgerEntries toWriteEntries = read.read(0,numberOfEntries - 1);
90+
Iterator<LedgerEntry> toTestIter = toTestEntries.iterator();
91+
Iterator<LedgerEntry> toWriteIter = toWriteEntries.iterator();
92+
while(toTestIter.hasNext()) {
93+
LedgerEntry toWriteEntry = toWriteIter.next();
94+
LedgerEntry toTestEntry = toTestIter.next();
95+
96+
assertEquals(toWriteEntry.getLedgerId(), toTestEntry.getLedgerId());
97+
assertEquals(toWriteEntry.getEntryId(), toTestEntry.getEntryId());
98+
assertEquals(toWriteEntry.getLength(), toTestEntry.getLength());
99+
assertEquals(toWriteEntry.getEntryBuffer(), toTestEntry.getEntryBuffer());
100+
}
101+
toTestEntries = toTest.read(1, numberOfEntries - 1);
102+
toWriteEntries = read.read(1,numberOfEntries - 1);
103+
toTestIter = toTestEntries.iterator();
104+
toWriteIter = toWriteEntries.iterator();
105+
while(toTestIter.hasNext()) {
106+
LedgerEntry toWriteEntry = toWriteIter.next();
107+
LedgerEntry toTestEntry = toTestIter.next();
108+
109+
assertEquals(toWriteEntry.getLedgerId(), toTestEntry.getLedgerId());
110+
assertEquals(toWriteEntry.getEntryId(), toTestEntry.getEntryId());
111+
assertEquals(toWriteEntry.getLength(), toTestEntry.getLength());
112+
assertEquals(toWriteEntry.getEntryBuffer(), toTestEntry.getEntryBuffer());
113+
}
114+
115+
// check the file located in the local file system
116+
Path offloadedFilePath = Paths.get(basePath, mlName);
117+
assertEquals(Files.exists(offloadedFilePath), true);
118+
}
119+
}

‎tiered-storage/file-system/src/test/resources/filesystem_offload_core_site.xml

+48
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
<!--
2+
3+
Licensed to the Apache Software Foundation (ASF) under one
4+
or more contributor license agreements. See the NOTICE file
5+
distributed with this work for additional information
6+
regarding copyright ownership. The ASF licenses this file
7+
to you under the Apache License, Version 2.0 (the
8+
"License"); you may not use this file except in compliance
9+
with the License. You may obtain a copy of the License at
10+
11+
http://www.apache.org/licenses/LICENSE-2.0
12+
13+
Unless required by applicable law or agreed to in writing,
14+
software distributed under the License is distributed on an
15+
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
16+
KIND, either express or implied. See the License for the
17+
specific language governing permissions and limitations
18+
under the License.
19+
20+
-->
21+
<configuration>
22+
<!--file system uri, necessary-->
23+
<property>
24+
<name>fs.defaultFS</name>
25+
<value></value>
26+
</property>
27+
<property>
28+
<name>hadoop.tmp.dir</name>
29+
<value>pulsar</value>
30+
</property>
31+
<property>
32+
<name>io.file.buffer.size</name>
33+
<value>4096</value>
34+
</property>
35+
<property>
36+
<name>io.seqfile.compress.blocksize</name>
37+
<value>1000000</value>
38+
</property>
39+
<property>
40+
<name>io.seqfile.compression.type</name>
41+
<value>BLOCK</value>
42+
</property>
43+
<property>
44+
<name>io.map.index.interval</name>
45+
<value>128</value>
46+
</property>
47+
48+
</configuration>

0 commit comments

Comments
 (0)