sijie closed pull request #2371: [tiered storage] move offloader implementation
to a separate module
URL: https://github.com/apache/incubator-pulsar/pull/2371
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/pom.xml b/pom.xml
index 2ce7e2fcb8..f60421cfc0 100644
--- a/pom.xml
+++ b/pom.xml
@@ -76,6 +76,7 @@ flexible messaging model and an intuitive client
API.</description>
<module>buildtools</module>
<module>managed-ledger</module>
<module>managed-ledger-shaded</module>
+ <module>tiered-storage</module>
<module>pulsar-common</module>
<module>pulsar-broker-common</module>
<module>pulsar-broker</module>
diff --git a/pulsar-broker/pom.xml b/pulsar-broker/pom.xml
index 8e5d3d3d7f..f3410b5863 100644
--- a/pulsar-broker/pom.xml
+++ b/pulsar-broker/pom.xml
@@ -97,6 +97,12 @@
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>tiered-storage-jcloud</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>pulsar-broker-common</artifactId>
@@ -125,11 +131,6 @@
<scope>test</scope>
</dependency>
- <dependency>
- <groupId>com.amazonaws</groupId>
- <artifactId>aws-java-sdk-core</artifactId>
- </dependency>
-
<!-- functions related dependencies (begin) -->
<dependency>
@@ -274,11 +275,6 @@
<scope>test</scope>
</dependency>
- <dependency>
- <groupId>org.apache.pulsar</groupId>
- <artifactId>jclouds-shaded</artifactId>
- <version>${project.version}</version>
- </dependency>
</dependencies>
<build>
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index ff224bcbbb..f6319aeaa7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -22,6 +22,7 @@
import static org.apache.pulsar.broker.admin.impl.NamespacesBase.getBundles;
import static
org.apache.pulsar.broker.cache.ConfigurationCacheService.POLICIES;
+import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
@@ -61,6 +62,7 @@
import org.apache.pulsar.broker.loadbalance.LoadSheddingTask;
import org.apache.pulsar.broker.loadbalance.impl.LoadManagerShared;
import org.apache.pulsar.broker.namespace.NamespaceService;
+import org.apache.pulsar.broker.offload.TieredStorageConfigurationData;
import org.apache.pulsar.broker.offload.impl.BlobStoreManagedLedgerOffloader;
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.Topic;
@@ -654,16 +656,51 @@ public LedgerOffloader getManagedLedgerOffloader() {
return offloader;
}
+ // TODO: improve the user metadata in subsequent changes
+ static final String METADATA_SOFTWARE_VERSION_KEY =
"S3ManagedLedgerOffloaderSoftwareVersion";
+ static final String METADATA_SOFTWARE_GITSHA_KEY =
"S3ManagedLedgerOffloaderSoftwareGitSha";
+
+
public synchronized LedgerOffloader
createManagedLedgerOffloader(ServiceConfiguration conf)
throws PulsarServerException {
if (conf.getManagedLedgerOffloadDriver() != null
&&
BlobStoreManagedLedgerOffloader.driverSupported(conf.getManagedLedgerOffloadDriver()))
{
- return BlobStoreManagedLedgerOffloader.create(conf,
getOffloaderScheduler(conf));
+ try {
+ return BlobStoreManagedLedgerOffloader.create(
+ getTieredStorageConf(conf),
+ ImmutableMap.of(
+ METADATA_SOFTWARE_VERSION_KEY.toLowerCase(),
PulsarBrokerVersionStringUtils.getNormalizedVersionString(),
+ METADATA_SOFTWARE_GITSHA_KEY.toLowerCase(),
PulsarBrokerVersionStringUtils.getGitSha()
+ ),
+ getOffloaderScheduler(conf));
+ } catch (IOException ioe) {
+ throw new PulsarServerException(ioe.getMessage(),
ioe.getCause());
+ }
} else {
return NullLedgerOffloader.INSTANCE;
}
}
+ private static TieredStorageConfigurationData
getTieredStorageConf(ServiceConfiguration serverConf) {
+ TieredStorageConfigurationData tsConf = new
TieredStorageConfigurationData();
+ // generic settings
+
tsConf.setManagedLedgerOffloadDriver(serverConf.getManagedLedgerOffloadDriver());
+
tsConf.setManagedLedgerOffloadMaxThreads(serverConf.getManagedLedgerOffloadMaxThreads());
+ // s3 settings
+
tsConf.setS3ManagedLedgerOffloadRegion(serverConf.getS3ManagedLedgerOffloadRegion());
+
tsConf.setS3ManagedLedgerOffloadBucket(serverConf.getS3ManagedLedgerOffloadBucket());
+
tsConf.setS3ManagedLedgerOffloadServiceEndpoint(serverConf.getS3ManagedLedgerOffloadServiceEndpoint());
+
tsConf.setS3ManagedLedgerOffloadMaxBlockSizeInBytes(serverConf.getS3ManagedLedgerOffloadMaxBlockSizeInBytes());
+
tsConf.setS3ManagedLedgerOffloadReadBufferSizeInBytes(serverConf.getS3ManagedLedgerOffloadReadBufferSizeInBytes());
+ // gcs settings
+
tsConf.setGcsManagedLedgerOffloadRegion(serverConf.getGcsManagedLedgerOffloadRegion());
+
tsConf.setGcsManagedLedgerOffloadBucket(serverConf.getGcsManagedLedgerOffloadBucket());
+
tsConf.setGcsManagedLedgerOffloadServiceAccountKeyFile(serverConf.getGcsManagedLedgerOffloadServiceAccountKeyFile());
+
tsConf.setGcsManagedLedgerOffloadMaxBlockSizeInBytes(serverConf.getGcsManagedLedgerOffloadMaxBlockSizeInBytes());
+
tsConf.setGcsManagedLedgerOffloadReadBufferSizeInBytes(serverConf.getGcsManagedLedgerOffloadReadBufferSizeInBytes());
+ return tsConf;
+ }
+
public ZooKeeperCache getLocalZkCache() {
return localZkCache;
}
diff --git a/tiered-storage/jcloud/pom.xml b/tiered-storage/jcloud/pom.xml
new file mode 100644
index 0000000000..4c4bc39aee
--- /dev/null
+++ b/tiered-storage/jcloud/pom.xml
@@ -0,0 +1,58 @@
+<!--
+
+ 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>tiered-storage-parent</artifactId>
+ <version>2.2.0-incubating-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>tiered-storage-jcloud</artifactId>
+ <name>Apache Pulsar :: Tiered Storage :: JCloud</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>managed-ledger-original</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>jclouds-shaded</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>com.amazonaws</groupId>
+ <artifactId>aws-java-sdk-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>managed-ledger-original</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+</project>
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/BackedInputStream.java
b/tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/BackedInputStream.java
similarity index 100%
rename from
pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/BackedInputStream.java
rename to
tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/BackedInputStream.java
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/BlockAwareSegmentInputStream.java
b/tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/BlockAwareSegmentInputStream.java
similarity index 100%
rename from
pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/BlockAwareSegmentInputStream.java
rename to
tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/BlockAwareSegmentInputStream.java
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/DataBlockHeader.java
b/tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/DataBlockHeader.java
similarity index 100%
rename from
pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/DataBlockHeader.java
rename to
tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/DataBlockHeader.java
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/OffloadIndexBlock.java
b/tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/OffloadIndexBlock.java
similarity index 100%
rename from
pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/OffloadIndexBlock.java
rename to
tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/OffloadIndexBlock.java
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/OffloadIndexBlockBuilder.java
b/tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/OffloadIndexBlockBuilder.java
similarity index 100%
rename from
pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/OffloadIndexBlockBuilder.java
rename to
tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/OffloadIndexBlockBuilder.java
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/OffloadIndexEntry.java
b/tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/OffloadIndexEntry.java
similarity index 100%
rename from
pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/OffloadIndexEntry.java
rename to
tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/OffloadIndexEntry.java
diff --git
a/tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/TieredStorageConfigurationData.java
b/tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/TieredStorageConfigurationData.java
new file mode 100644
index 0000000000..c7ef3fd322
--- /dev/null
+++
b/tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/TieredStorageConfigurationData.java
@@ -0,0 +1,69 @@
+/**
+ * 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.offload;
+
+import java.io.Serializable;
+import lombok.Data;
+
+/**
+ * Configuration for tiered storage.
+ */
+@Data
+public class TieredStorageConfigurationData implements Serializable, Cloneable{
+
+ /**** --- Ledger Offloading --- ****/
+ // Driver to use to offload old data to long term storage
+ private String managedLedgerOffloadDriver = null;
+
+ // Maximum number of thread pool threads for ledger offloading
+ private int managedLedgerOffloadMaxThreads = 2;
+
+ // For Amazon S3 ledger offload, AWS region
+ private String s3ManagedLedgerOffloadRegion = null;
+
+ // For Amazon S3 ledger offload, Bucket to place offloaded ledger into
+ private String s3ManagedLedgerOffloadBucket = null;
+
+ // For Amazon S3 ledger offload, Alternative endpoint to connect to
(useful for testing)
+ private String s3ManagedLedgerOffloadServiceEndpoint = null;
+
+ // For Amazon S3 ledger offload, Max block size in bytes.
+ private int s3ManagedLedgerOffloadMaxBlockSizeInBytes = 64 * 1024 * 1024;
// 64MB
+
+ // For Amazon S3 ledger offload, Read buffer size in bytes.
+ private int s3ManagedLedgerOffloadReadBufferSizeInBytes = 1024 * 1024; //
1MB
+
+ // For Google Cloud Storage ledger offload, region where offload bucket is
located.
+ // reference this page for more details:
https://cloud.google.com/storage/docs/bucket-locations
+ private String gcsManagedLedgerOffloadRegion = null;
+
+ // For Google Cloud Storage ledger offload, Bucket to place offloaded
ledger into
+ private String gcsManagedLedgerOffloadBucket = null;
+
+ // For Google Cloud Storage ledger offload, Max block size in bytes.
+ private int gcsManagedLedgerOffloadMaxBlockSizeInBytes = 64 * 1024 * 1024;
// 64MB
+
+ // For Google Cloud Storage ledger offload, Read buffer size in bytes.
+ private int gcsManagedLedgerOffloadReadBufferSizeInBytes = 1024 * 1024; //
1MB
+
+ // For Google Cloud Storage, path to json file containing service account
credentials.
+ // For more details, see the "Service Accounts" section of
https://support.google.com/googleapi/answer/6158849
+ private String gcsManagedLedgerOffloadServiceAccountKeyFile = null;
+
+}
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/impl/BlobStoreBackedInputStreamImpl.java
b/tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/impl/BlobStoreBackedInputStreamImpl.java
similarity index 100%
rename from
pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/impl/BlobStoreBackedInputStreamImpl.java
rename to
tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/impl/BlobStoreBackedInputStreamImpl.java
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/impl/BlobStoreBackedReadHandleImpl.java
b/tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/impl/BlobStoreBackedReadHandleImpl.java
similarity index 98%
rename from
pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/impl/BlobStoreBackedReadHandleImpl.java
rename to
tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/impl/BlobStoreBackedReadHandleImpl.java
index 36b382be96..7598d58471 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/impl/BlobStoreBackedReadHandleImpl.java
+++
b/tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/impl/BlobStoreBackedReadHandleImpl.java
@@ -18,7 +18,6 @@
*/
package org.apache.pulsar.broker.offload.impl;
-import com.amazonaws.AmazonClientException;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import java.io.DataInputStream;
@@ -186,7 +185,7 @@ public static ReadHandle open(ScheduledExecutorService
executor,
BlobStore blobStore, String bucket, String
key, String indexKey,
VersionCheck versionCheck,
long ledgerId, int readBufferSize)
- throws AmazonClientException, IOException {
+ throws IOException {
Blob blob = blobStore.getBlob(bucket, indexKey);
versionCheck.check(indexKey, blob);
OffloadIndexBlockBuilder indexBuilder =
OffloadIndexBlockBuilder.create();
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/impl/BlobStoreManagedLedgerOffloader.java
b/tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/impl/BlobStoreManagedLedgerOffloader.java
similarity index 86%
rename from
pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/impl/BlobStoreManagedLedgerOffloader.java
rename to
tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/impl/BlobStoreManagedLedgerOffloader.java
index fa0647c31e..b0345fae35 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/impl/BlobStoreManagedLedgerOffloader.java
+++
b/tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/impl/BlobStoreManagedLedgerOffloader.java
@@ -25,6 +25,7 @@
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
import com.google.common.io.Files;
import java.io.File;
import java.io.IOException;
@@ -38,12 +39,10 @@
import org.apache.bookkeeper.client.api.ReadHandle;
import org.apache.bookkeeper.common.util.OrderedScheduler;
import org.apache.bookkeeper.mledger.LedgerOffloader;
-import org.apache.pulsar.broker.PulsarServerException;
-import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.offload.BlockAwareSegmentInputStream;
import org.apache.pulsar.broker.offload.OffloadIndexBlock;
import org.apache.pulsar.broker.offload.OffloadIndexBlockBuilder;
-import org.apache.pulsar.utils.PulsarBrokerVersionStringUtils;
+import org.apache.pulsar.broker.offload.TieredStorageConfigurationData;
import org.jclouds.Constants;
import org.jclouds.ContextBuilder;
import org.jclouds.blobstore.BlobStore;
@@ -71,8 +70,6 @@
// use these keys for both s3 and gcs.
static final String METADATA_FORMAT_VERSION_KEY =
"S3ManagedLedgerOffloaderFormatVersion";
- static final String METADATA_SOFTWARE_VERSION_KEY =
"S3ManagedLedgerOffloaderSoftwareVersion";
- static final String METADATA_SOFTWARE_GITSHA_KEY =
"S3ManagedLedgerOffloaderSoftwareGitSha";
static final String CURRENT_VERSION = String.valueOf(1);
public static boolean driverSupported(String driver) {
@@ -87,11 +84,11 @@ public static boolean isGcsDriver(String driver) {
return driver.equalsIgnoreCase(DRIVER_NAMES[2]);
}
- private static void addVersionInfo(BlobBuilder blobBuilder) {
- blobBuilder.userMetadata(ImmutableMap.of(
- METADATA_FORMAT_VERSION_KEY.toLowerCase(), CURRENT_VERSION,
- METADATA_SOFTWARE_VERSION_KEY.toLowerCase(),
PulsarBrokerVersionStringUtils.getNormalizedVersionString(),
- METADATA_SOFTWARE_GITSHA_KEY.toLowerCase(),
PulsarBrokerVersionStringUtils.getGitSha()));
+ private static void addVersionInfo(BlobBuilder blobBuilder, Map<String,
String> userMetadata) {
+ ImmutableMap.Builder<String, String> metadataBuilder =
ImmutableMap.builder();
+ metadataBuilder.putAll(userMetadata);
+ metadataBuilder.put(METADATA_FORMAT_VERSION_KEY.toLowerCase(),
CURRENT_VERSION);
+ blobBuilder.userMetadata(metadataBuilder.build());
}
private final VersionCheck VERSION_CHECK = (key, blob) -> {
@@ -99,7 +96,7 @@ private static void addVersionInfo(BlobBuilder blobBuilder) {
String version =
blob.getMetadata().getUserMetadata().get(METADATA_FORMAT_VERSION_KEY.toLowerCase());
if (version == null || !version.equals(CURRENT_VERSION)) {
throw new IOException(String.format("Invalid object version %s for
%s, expect %s",
- version, key,
CURRENT_VERSION));
+ version, key, CURRENT_VERSION));
}
};
@@ -114,13 +111,21 @@ private static void addVersionInfo(BlobBuilder
blobBuilder) {
private BlobStoreContext context;
private BlobStore blobStore;
Location location = null;
+ private final Map<String, String> userMetadata;
- public static BlobStoreManagedLedgerOffloader create(ServiceConfiguration
conf,
+ @VisibleForTesting
+ static BlobStoreManagedLedgerOffloader
create(TieredStorageConfigurationData conf,
+ OrderedScheduler scheduler)
throws IOException {
+ return create(conf, Maps.newHashMap(), scheduler);
+ }
+
+ public static BlobStoreManagedLedgerOffloader
create(TieredStorageConfigurationData conf,
+ Map<String, String>
userMetadata,
OrderedScheduler
scheduler)
- throws PulsarServerException {
+ throws IOException {
String driver = conf.getManagedLedgerOffloadDriver();
if (!driverSupported(driver)) {
- throw new PulsarServerException(
+ throw new IOException(
"Not support this kind of driver as offload backend: " +
driver);
}
@@ -139,26 +144,27 @@ public static BlobStoreManagedLedgerOffloader
create(ServiceConfiguration conf,
conf.getGcsManagedLedgerOffloadReadBufferSizeInBytes();
if (isS3Driver(driver) && Strings.isNullOrEmpty(region) &&
Strings.isNullOrEmpty(endpoint)) {
- throw new PulsarServerException(
+ throw new IOException(
"Either s3ManagedLedgerOffloadRegion or
s3ManagedLedgerOffloadServiceEndpoint must be set"
+ " if s3 offload enabled");
}
if (Strings.isNullOrEmpty(bucket)) {
- throw new PulsarServerException(
+ throw new IOException(
"ManagedLedgerOffloadBucket cannot be empty for s3 and gcs
offload");
}
if (maxBlockSize < 5*1024*1024) {
- throw new PulsarServerException(
+ throw new IOException(
"ManagedLedgerOffloadMaxBlockSizeInBytes cannot be less than
5MB for s3 and gcs offload");
}
Credentials credentials = getCredentials(driver, conf);
- return new BlobStoreManagedLedgerOffloader(driver, bucket, scheduler,
maxBlockSize, readBufferSize, endpoint, region, credentials);
+ return new BlobStoreManagedLedgerOffloader(driver, bucket, scheduler,
+ maxBlockSize, readBufferSize, endpoint, region, credentials,
userMetadata);
}
- public static Credentials getCredentials(String driver,
ServiceConfiguration conf) throws PulsarServerException {
+ public static Credentials getCredentials(String driver,
TieredStorageConfigurationData conf) throws IOException {
// credentials:
// for s3, get by DefaultAWSCredentialsProviderChain.
// for gcs, use downloaded file 'google_creds.json', which contains
service account key by
@@ -167,7 +173,7 @@ public static Credentials getCredentials(String driver,
ServiceConfiguration con
if (isGcsDriver(driver)) {
String gcsKeyPath =
conf.getGcsManagedLedgerOffloadServiceAccountKeyFile();
if (Strings.isNullOrEmpty(gcsKeyPath)) {
- throw new PulsarServerException(
+ throw new IOException(
"The service account key path is empty for GCS driver");
}
try {
@@ -175,7 +181,7 @@ public static Credentials getCredentials(String driver,
ServiceConfiguration con
return new GoogleCredentialsFromJson(gcsKeyContent).get();
} catch (IOException ioe) {
log.error("Cannot read GCS service account credentials file:
{}", gcsKeyPath);
- throw new PulsarServerException(ioe);
+ throw new IOException(ioe);
}
} else if (isS3Driver(driver)) {
AWSCredentials credentials = null;
@@ -195,19 +201,27 @@ public static Credentials getCredentials(String driver,
ServiceConfiguration con
}
return new Credentials(id, key);
} else {
- throw new PulsarServerException(
+ throw new IOException(
"Not support this kind of driver: " + driver);
}
}
+
// build context for jclouds BlobStoreContext
BlobStoreManagedLedgerOffloader(String driver, String container,
OrderedScheduler scheduler,
- int maxBlockSize, int readBufferSize, String
endpoint, String region, Credentials credentials) {
+ int maxBlockSize, int readBufferSize,
String endpoint, String region, Credentials credentials) {
+ this(driver, container, scheduler, maxBlockSize, readBufferSize,
endpoint, region, credentials, Maps.newHashMap());
+ }
+
+ BlobStoreManagedLedgerOffloader(String driver, String container,
OrderedScheduler scheduler,
+ int maxBlockSize, int readBufferSize,
+ String endpoint, String region,
Credentials credentials,
+ Map<String, String> userMetadata) {
this.scheduler = scheduler;
this.readBufferSize = readBufferSize;
-
this.bucket = container;
this.maxBlockSize = maxBlockSize;
+ this.userMetadata = userMetadata;
Properties overrides = new Properties();
// This property controls the number of parts being uploaded in
parallel.
@@ -238,11 +252,18 @@ public static Credentials getCredentials(String driver,
ServiceConfiguration con
@VisibleForTesting
BlobStoreManagedLedgerOffloader(BlobStore blobStore, String container,
OrderedScheduler scheduler,
int maxBlockSize, int readBufferSize) {
+ this(blobStore, container, scheduler, maxBlockSize, readBufferSize,
Maps.newHashMap());
+ }
+
+ BlobStoreManagedLedgerOffloader(BlobStore blobStore, String container,
OrderedScheduler scheduler,
+ int maxBlockSize, int readBufferSize,
+ Map<String, String> userMetadata) {
this.scheduler = scheduler;
this.readBufferSize = readBufferSize;
this.bucket = container;
this.maxBlockSize = maxBlockSize;
this.blobStore = blobStore;
+ this.userMetadata = userMetadata;
}
static String dataBlockOffloadKey(long ledgerId, UUID uuid) {
@@ -285,7 +306,7 @@ public void deleteBucket() {
// init multi part upload for data block.
try {
BlobBuilder blobBuilder = blobStore.blobBuilder(dataBlockKey);
- addVersionInfo(blobBuilder);
+ addVersionInfo(blobBuilder, userMetadata);
Blob blob = blobBuilder.build();
mpu = blobStore.initiateMultipartUpload(bucket,
blob.getMetadata(), new PutOptions());
} catch (Throwable t) {
@@ -348,7 +369,7 @@ public void deleteBucket() {
OffloadIndexBlock.IndexInputStream indexStream =
index.toStream()) {
// write the index block
BlobBuilder blobBuilder = blobStore.blobBuilder(indexBlockKey);
- addVersionInfo(blobBuilder);
+ addVersionInfo(blobBuilder, userMetadata);
Payload indexPayload =
Payloads.newInputStreamPayload(indexStream);
indexPayload.getContentMetadata().setContentLength((long)indexStream.getStreamSize());
indexPayload.getContentMetadata().setContentType("application/octet-stream");
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/impl/BlockAwareSegmentInputStreamImpl.java
b/tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/impl/BlockAwareSegmentInputStreamImpl.java
similarity index 100%
rename from
pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/impl/BlockAwareSegmentInputStreamImpl.java
rename to
tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/impl/BlockAwareSegmentInputStreamImpl.java
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/impl/DataBlockHeaderImpl.java
b/tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/impl/DataBlockHeaderImpl.java
similarity index 100%
rename from
pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/impl/DataBlockHeaderImpl.java
rename to
tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/impl/DataBlockHeaderImpl.java
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/impl/OffloadIndexBlockBuilderImpl.java
b/tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/impl/OffloadIndexBlockBuilderImpl.java
similarity index 100%
rename from
pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/impl/OffloadIndexBlockBuilderImpl.java
rename to
tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/impl/OffloadIndexBlockBuilderImpl.java
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/impl/OffloadIndexBlockImpl.java
b/tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/impl/OffloadIndexBlockImpl.java
similarity index 100%
rename from
pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/impl/OffloadIndexBlockImpl.java
rename to
tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/impl/OffloadIndexBlockImpl.java
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/impl/OffloadIndexEntryImpl.java
b/tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/impl/OffloadIndexEntryImpl.java
similarity index 100%
rename from
pulsar-broker/src/main/java/org/apache/pulsar/broker/offload/impl/OffloadIndexEntryImpl.java
rename to
tiered-storage/jcloud/src/main/java/org/apache/pulsar/broker/offload/impl/OffloadIndexEntryImpl.java
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/offload/BlobStoreBackedInputStreamTest.java
b/tiered-storage/jcloud/src/test/java/org/apache/pulsar/broker/offload/BlobStoreBackedInputStreamTest.java
similarity index 100%
rename from
pulsar-broker/src/test/java/org/apache/pulsar/broker/offload/BlobStoreBackedInputStreamTest.java
rename to
tiered-storage/jcloud/src/test/java/org/apache/pulsar/broker/offload/BlobStoreBackedInputStreamTest.java
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/offload/BlobStoreTestBase.java
b/tiered-storage/jcloud/src/test/java/org/apache/pulsar/broker/offload/BlobStoreTestBase.java
similarity index 100%
rename from
pulsar-broker/src/test/java/org/apache/pulsar/broker/offload/BlobStoreTestBase.java
rename to
tiered-storage/jcloud/src/test/java/org/apache/pulsar/broker/offload/BlobStoreTestBase.java
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/offload/impl/BlobStoreManagedLedgerOffloaderTest.java
b/tiered-storage/jcloud/src/test/java/org/apache/pulsar/broker/offload/impl/BlobStoreManagedLedgerOffloaderTest.java
similarity index 93%
rename from
pulsar-broker/src/test/java/org/apache/pulsar/broker/offload/impl/BlobStoreManagedLedgerOffloaderTest.java
rename to
tiered-storage/jcloud/src/test/java/org/apache/pulsar/broker/offload/impl/BlobStoreManagedLedgerOffloaderTest.java
index f1e43c2144..0999d63768 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/offload/impl/BlobStoreManagedLedgerOffloaderTest.java
+++
b/tiered-storage/jcloud/src/test/java/org/apache/pulsar/broker/offload/impl/BlobStoreManagedLedgerOffloaderTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker.offload.impl;
+import static java.nio.charset.StandardCharsets.UTF_8;
import static
org.apache.pulsar.broker.offload.impl.BlobStoreManagedLedgerOffloader.dataBlockOffloadKey;
import static
org.apache.pulsar.broker.offload.impl.BlobStoreManagedLedgerOffloader.indexBlockOffloadKey;
import static org.mockito.AdditionalAnswers.delegatesTo;
@@ -26,17 +27,19 @@
import static org.mockito.Matchers.anyLong;
import static org.mockito.Mockito.mock;
+import com.google.common.util.concurrent.MoreExecutors;
import java.io.File;
import java.io.IOException;
import java.lang.reflect.Method;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
+import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
-import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.LedgerHandle;
@@ -48,10 +51,12 @@
import org.apache.bookkeeper.client.api.ReadHandle;
import org.apache.bookkeeper.common.util.OrderedScheduler;
import org.apache.bookkeeper.mledger.LedgerOffloader;
-import org.apache.pulsar.broker.PulsarServerException;
-import org.apache.pulsar.broker.ServiceConfiguration;
-import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.bookkeeper.util.ZkUtils;
import org.apache.pulsar.broker.offload.BlobStoreTestBase;
+import org.apache.pulsar.broker.offload.TieredStorageConfigurationData;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.MockZooKeeper;
+import org.apache.zookeeper.data.ACL;
import org.jclouds.blobstore.BlobStore;
import org.jclouds.blobstore.options.CopyOptions;
import org.mockito.Mockito;
@@ -61,10 +66,21 @@
import org.testng.annotations.Test;
import org.testng.collections.Maps;
-@Slf4j
class BlobStoreManagedLedgerOffloaderTest extends BlobStoreTestBase {
private static final Logger log =
LoggerFactory.getLogger(BlobStoreManagedLedgerOffloaderTest.class);
+ private static MockZooKeeper createMockZooKeeper() throws Exception {
+ MockZooKeeper zk =
MockZooKeeper.newInstance(MoreExecutors.newDirectExecutorService());
+ List<ACL> dummyAclList = new ArrayList<ACL>(0);
+
+ ZkUtils.createFullPathOptimistic(zk, "/ledgers/available/192.168.1.1:"
+ 5000,
+ "".getBytes(UTF_8), dummyAclList, CreateMode.PERSISTENT);
+
+ zk.create("/ledgers/LAYOUT", "1\nflat:1".getBytes(UTF_8), dummyAclList,
+ CreateMode.PERSISTENT);
+ return zk;
+ }
+
private static final int DEFAULT_BLOCK_SIZE = 5*1024*1024;
private static final int DEFAULT_READ_BUFFER_SIZE = 1*1024*1024;
final OrderedScheduler scheduler;
@@ -72,7 +88,7 @@
BlobStoreManagedLedgerOffloaderTest() throws Exception {
scheduler =
OrderedScheduler.newSchedulerBuilder().numThreads(1).name("offloader").build();
- bk = new
MockBookKeeper(MockedPulsarServiceBaseTest.createMockZooKeeper());
+ bk = new MockBookKeeper(createMockZooKeeper());
}
private ReadHandle buildReadHandle() throws Exception {
@@ -139,35 +155,35 @@ public void testBucketDoesNotExist() throws Exception {
@Test
public void testNoRegionConfigured() throws Exception {
- ServiceConfiguration conf = new ServiceConfiguration();
+ TieredStorageConfigurationData conf = new
TieredStorageConfigurationData();
conf.setManagedLedgerOffloadDriver("s3");
conf.setS3ManagedLedgerOffloadBucket(BUCKET);
try {
BlobStoreManagedLedgerOffloader.create(conf, scheduler);
Assert.fail("Should have thrown exception");
- } catch (PulsarServerException pse) {
+ } catch (IOException pse) {
// correct
}
}
@Test
public void testNoBucketConfigured() throws Exception {
- ServiceConfiguration conf = new ServiceConfiguration();
+ TieredStorageConfigurationData conf = new
TieredStorageConfigurationData();
conf.setManagedLedgerOffloadDriver("s3");
conf.setS3ManagedLedgerOffloadRegion("eu-west-1");
try {
BlobStoreManagedLedgerOffloader.create(conf, scheduler);
Assert.fail("Should have thrown exception");
- } catch (PulsarServerException pse) {
+ } catch (IOException pse) {
// correct
}
}
@Test
public void testSmallBlockSizeConfigured() throws Exception {
- ServiceConfiguration conf = new ServiceConfiguration();
+ TieredStorageConfigurationData conf = new
TieredStorageConfigurationData();
conf.setManagedLedgerOffloadDriver("s3");
conf.setS3ManagedLedgerOffloadRegion("eu-west-1");
conf.setS3ManagedLedgerOffloadBucket(BUCKET);
@@ -176,21 +192,21 @@ public void testSmallBlockSizeConfigured() throws
Exception {
try {
BlobStoreManagedLedgerOffloader.create(conf, scheduler);
Assert.fail("Should have thrown exception");
- } catch (PulsarServerException pse) {
+ } catch (IOException pse) {
// correct
}
}
@Test
public void testGcsNoKeyPath() throws Exception {
- ServiceConfiguration conf = new ServiceConfiguration();
+ TieredStorageConfigurationData conf = new
TieredStorageConfigurationData();
conf.setManagedLedgerOffloadDriver("google-cloud-storage");
conf.setGcsManagedLedgerOffloadBucket(BUCKET);
try {
BlobStoreManagedLedgerOffloader.create(conf, scheduler);
Assert.fail("Should have thrown exception");
- } catch (PulsarServerException pse) {
+ } catch (IOException pse) {
// correct
log.error("Expected pse", pse);
}
@@ -198,7 +214,7 @@ public void testGcsNoKeyPath() throws Exception {
@Test
public void testGcsNoBucketConfigured() throws Exception {
- ServiceConfiguration conf = new ServiceConfiguration();
+ TieredStorageConfigurationData conf = new
TieredStorageConfigurationData();
conf.setManagedLedgerOffloadDriver("google-cloud-storage");
File tmpKeyFile = File.createTempFile("gcsOffload", "json");
conf.setGcsManagedLedgerOffloadServiceAccountKeyFile(tmpKeyFile.getAbsolutePath());
@@ -206,7 +222,7 @@ public void testGcsNoBucketConfigured() throws Exception {
try {
BlobStoreManagedLedgerOffloader.create(conf, scheduler);
Assert.fail("Should have thrown exception");
- } catch (PulsarServerException pse) {
+ } catch (IOException pse) {
// correct
log.error("Expected pse", pse);
}
@@ -214,7 +230,7 @@ public void testGcsNoBucketConfigured() throws Exception {
@Test
public void testGcsSmallBlockSizeConfigured() throws Exception {
- ServiceConfiguration conf = new ServiceConfiguration();
+ TieredStorageConfigurationData conf = new
TieredStorageConfigurationData();
conf.setManagedLedgerOffloadDriver("google-cloud-storage");
File tmpKeyFile = File.createTempFile("gcsOffload", "json");
conf.setGcsManagedLedgerOffloadServiceAccountKeyFile(tmpKeyFile.getAbsolutePath());
@@ -224,7 +240,7 @@ public void testGcsSmallBlockSizeConfigured() throws
Exception {
try {
BlobStoreManagedLedgerOffloader.create(conf, scheduler);
Assert.fail("Should have thrown exception");
- } catch (PulsarServerException pse) {
+ } catch (IOException pse) {
// correct
log.error("Expected pse", pse);
}
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/offload/impl/BlockAwareSegmentInputStreamTest.java
b/tiered-storage/jcloud/src/test/java/org/apache/pulsar/broker/offload/impl/BlockAwareSegmentInputStreamTest.java
similarity index 100%
rename from
pulsar-broker/src/test/java/org/apache/pulsar/broker/offload/impl/BlockAwareSegmentInputStreamTest.java
rename to
tiered-storage/jcloud/src/test/java/org/apache/pulsar/broker/offload/impl/BlockAwareSegmentInputStreamTest.java
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/offload/impl/DataBlockHeaderTest.java
b/tiered-storage/jcloud/src/test/java/org/apache/pulsar/broker/offload/impl/DataBlockHeaderTest.java
similarity index 100%
rename from
pulsar-broker/src/test/java/org/apache/pulsar/broker/offload/impl/DataBlockHeaderTest.java
rename to
tiered-storage/jcloud/src/test/java/org/apache/pulsar/broker/offload/impl/DataBlockHeaderTest.java
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/offload/impl/OffloadIndexTest.java
b/tiered-storage/jcloud/src/test/java/org/apache/pulsar/broker/offload/impl/OffloadIndexTest.java
similarity index 100%
rename from
pulsar-broker/src/test/java/org/apache/pulsar/broker/offload/impl/OffloadIndexTest.java
rename to
tiered-storage/jcloud/src/test/java/org/apache/pulsar/broker/offload/impl/OffloadIndexTest.java
diff --git a/tiered-storage/pom.xml b/tiered-storage/pom.xml
new file mode 100644
index 0000000000..d8f25b8c80
--- /dev/null
+++ b/tiered-storage/pom.xml
@@ -0,0 +1,38 @@
+<!--
+
+ 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+ <packaging>pom</packaging>
+ <parent>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar</artifactId>
+ <version>2.2.0-incubating-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>tiered-storage-parent</artifactId>
+ <name>Apache Pulsar :: Tiered Storage :: Parent</name>
+
+ <modules>
+ <module>jcloud</module>
+ </modules>
+</project>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
