You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by ap...@apache.org on 2019/10/02 20:20:43 UTC
[hbase] branch branch-1.3 updated: HBASE-23101 Backport HBASE-22380
to branch-1
This is an automated email from the ASF dual-hosted git repository.
apurtell pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/hbase.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 84f95fc HBASE-23101 Backport HBASE-22380 to branch-1
84f95fc is described below
commit 84f95fc18c6ff7fdc4bf0a39eaa23c94079be643
Author: Wellington Chevreuil <wc...@apache.org>
AuthorDate: Wed Oct 2 17:49:14 2019 +0100
HBASE-23101 Backport HBASE-22380 to branch-1
Fixes #680
Signed-off-by: Andrew Purtell <ap...@apache.org>
Conflicts:
hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/ClientProtos.java
hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/SecureBulkLoadProtos.java
hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/WALProtos.java
hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/LoadIncrementalHFiles.java
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java
---
.../client/coprocessor/SecureBulkLoadClient.java | 21 +-
.../apache/hadoop/hbase/protobuf/ProtobufUtil.java | 60 +++-
.../hadoop/hbase/protobuf/RequestConverter.java | 23 +-
.../hbase/protobuf/generated/ClientProtos.java | 324 ++++++++++++++++-----
.../protobuf/generated/SecureBulkLoadProtos.java | 232 +++++++++++++--
.../hadoop/hbase/protobuf/generated/WALProtos.java | 222 +++++++++++++-
hbase-protocol/src/main/protobuf/Client.proto | 1 +
.../src/main/protobuf/SecureBulkLoad.proto | 1 +
hbase-protocol/src/main/protobuf/WAL.proto | 1 +
.../hbase/mapreduce/LoadIncrementalHFiles.java | 9 +-
.../apache/hadoop/hbase/regionserver/HRegion.java | 4 +-
.../hadoop/hbase/regionserver/HRegionServer.java | 2 +-
.../hadoop/hbase/regionserver/RSRpcServices.java | 19 +-
.../apache/hadoop/hbase/regionserver/Region.java | 5 +-
.../replication/regionserver/HFileReplicator.java | 5 +-
.../replication/regionserver/ReplicationSink.java | 46 ++-
.../security/access/SecureBulkLoadEndpoint.java | 4 +-
.../hadoop/hbase/regionserver/TestBulkLoad.java | 18 +-
.../regionserver/TestBulkLoadReplication.java | 298 +++++++++++++++++++
.../regionserver/TestHRegionReplayEvents.java | 2 +-
.../hbase/regionserver/wal/TestWALReplay.java | 4 +-
21 files changed, 1150 insertions(+), 151 deletions(-)
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/coprocessor/SecureBulkLoadClient.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/coprocessor/SecureBulkLoadClient.java
index 05db8d4..bfbd8dd 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/coprocessor/SecureBulkLoadClient.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/coprocessor/SecureBulkLoadClient.java
@@ -117,10 +117,24 @@ public class SecureBulkLoadClient {
}
}
+ /**
+ * @deprecated
+ * @param familyPaths
+ * @param userToken
+ * @param bulkToken
+ * @param startRow
+ * @return
+ * @throws IOException
+ */
public boolean bulkLoadHFiles(final List<Pair<byte[], String>> familyPaths,
- final Token<?> userToken,
- final String bulkToken,
- final byte[] startRow) throws IOException {
+ final Token<?> userToken, final String bulkToken,
+ final byte[] startRow) throws IOException {
+ return this.bulkLoadHFiles(familyPaths, userToken, bulkToken, startRow, null);
+ }
+
+ public boolean bulkLoadHFiles(final List<Pair<byte[], String>> familyPaths,
+ final Token<?> userToken, final String bulkToken,
+ final byte[] startRow, List<String> clusterIds) throws IOException {
// we never want to send a batch of HFiles to all regions, thus cannot call
// HTable#coprocessorService methods that take start and end rowkeys; see HBASE-9639
try {
@@ -151,6 +165,7 @@ public class SecureBulkLoadClient {
SecureBulkLoadProtos.SecureBulkLoadHFilesRequest.newBuilder()
.setFsToken(protoDT)
.addAllFamilyPath(protoFamilyPaths)
+ .addAllClusterIds(clusterIds != null ? clusterIds : new ArrayList<String>())
.setBulkToken(bulkToken).build();
ServerRpcController controller = new ServerRpcController();
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/protobuf/ProtobufUtil.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/protobuf/ProtobufUtil.java
index 5abc1b7..fec3766 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/protobuf/ProtobufUtil.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/protobuf/ProtobufUtil.java
@@ -1677,12 +1677,31 @@ public final class ProtobufUtil {
* @param assignSeqNum
* @return true if all are loaded
* @throws IOException
+ * @deprecated use bulkLoadHFile(final ClientService.BlockingInterface client,
+ * final List<Pair<byte[], String>> familyPaths, final byte[] regionName, boolean assignSeqNum,
+ * List<String> clusterIds) instead.
*/
public static boolean bulkLoadHFile(final ClientService.BlockingInterface client,
final List<Pair<byte[], String>> familyPaths,
final byte[] regionName, boolean assignSeqNum) throws IOException {
+ return bulkLoadHFile(client, familyPaths, regionName, assignSeqNum, null);
+ }
+
+ /**
+ * A helper to bulk load a list of HFiles using client protocol.
+ *
+ * @param client
+ * @param familyPaths
+ * @param regionName
+ * @param assignSeqNum
+ * @return true if all are loaded
+ * @throws IOException
+ */
+ public static boolean bulkLoadHFile(final ClientService.BlockingInterface client,
+ final List<Pair<byte[], String>> familyPaths,
+ final byte[] regionName, boolean assignSeqNum, List<String> clusterIds) throws IOException {
BulkLoadHFileRequest request =
- RequestConverter.buildBulkLoadHFileRequest(familyPaths, regionName, assignSeqNum);
+ RequestConverter.buildBulkLoadHFileRequest(familyPaths, regionName, assignSeqNum, clusterIds);
try {
BulkLoadHFileResponse response =
client.bulkLoadHFile(null, request);
@@ -3210,6 +3229,10 @@ public final class ProtobufUtil {
* Generates a marker for the WAL so that we propagate the notion of a bulk region load
* throughout the WAL.
*
+ * @deprecated use toBulkLoadDescriptor(TableName tableName, ByteString encodedRegionName,
+ * Map<byte[], List<Path>> storeFiles, Map<String, Long> storeFilesSize, long bulkloadSeqId,
+ * List<String> clusterIds) instead.
+ *
* @param tableName The tableName into which the bulk load is being imported into.
* @param encodedRegionName Encoded region name of the region which is being bulk loaded.
* @param storeFiles A set of store files of a column family are bulk loaded.
@@ -3219,17 +3242,42 @@ public final class ProtobufUtil {
* @return The WAL log marker for bulk loads.
*/
public static WALProtos.BulkLoadDescriptor toBulkLoadDescriptor(TableName tableName,
- ByteString encodedRegionName, Map<byte[], List<Path>> storeFiles,
- Map<String, Long> storeFilesSize, long bulkloadSeqId) {
+ ByteString encodedRegionName, Map<byte[], List<Path>> storeFiles,
+ Map<String, Long> storeFilesSize, long bulkloadSeqId) {
+ return toBulkLoadDescriptor(tableName, encodedRegionName, storeFiles,
+ storeFilesSize, bulkloadSeqId, null);
+ }
+
+ /**
+ * Generates a marker for the WAL so that we propagate the notion of a bulk region load
+ * throughout the WAL, keeping track of clusters who already applied the bulk event via
+ * the passed clusterIds parameter.
+ *
+ * @param tableName The tableName into which the bulk load is being imported into.
+ * @param encodedRegionName Encoded region name of the region which is being bulk loaded.
+ * @param storeFiles A set of store files of a column family are bulk loaded.
+ * @param storeFilesSize Map of store files and their lengths
+ * @param bulkloadSeqId sequence ID (by a force flush) used to create bulk load hfile name
+ * @param clusterIds The list of cluster Ids with the clusters where the bulk even had
+ * already been processed.
+ *
+ * @return The WAL log marker for bulk loads.
+ */
+ public static WALProtos.BulkLoadDescriptor toBulkLoadDescriptor(TableName tableName,
+ ByteString encodedRegionName, Map<byte[], List<Path>> storeFiles,
+ Map<String, Long> storeFilesSize, long bulkloadSeqId, List<String> clusterIds) {
BulkLoadDescriptor.Builder desc =
- BulkLoadDescriptor.newBuilder()
+ BulkLoadDescriptor.newBuilder()
.setTableName(ProtobufUtil.toProtoTableName(tableName))
.setEncodedRegionName(encodedRegionName).setBulkloadSeqNum(bulkloadSeqId);
+ if(clusterIds != null) {
+ desc.addAllClusterIds(clusterIds);
+ }
for (Map.Entry<byte[], List<Path>> entry : storeFiles.entrySet()) {
WALProtos.StoreDescriptor.Builder builder = StoreDescriptor.newBuilder()
- .setFamilyName(ByteStringer.wrap(entry.getKey()))
- .setStoreHomeDir(Bytes.toString(entry.getKey())); // relative to region
+ .setFamilyName(ByteStringer.wrap(entry.getKey()))
+ .setStoreHomeDir(Bytes.toString(entry.getKey())); // relative to region
for (Path path : entry.getValue()) {
String name = path.getName();
builder.addStoreFile(name);
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/protobuf/RequestConverter.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/protobuf/RequestConverter.java
index a2d665b..dd7a634 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/protobuf/RequestConverter.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/protobuf/RequestConverter.java
@@ -548,14 +548,32 @@ public final class RequestConverter {
/**
* Create a protocol buffer bulk load request
*
+ * @deprecated use buildBulkLoadHFileRequest(final List<Pair<byte[], String>> familyPaths,
+ * final byte[] regionName, boolean assignSeqNum, List<String> clusterIds)
+ *
* @param familyPaths
* @param regionName
* @param assignSeqNum
* @return a bulk load request
*/
public static BulkLoadHFileRequest buildBulkLoadHFileRequest(
+ final List<Pair<byte[], String>> familyPaths,
+ final byte[] regionName, boolean assignSeqNum) {
+ return buildBulkLoadHFileRequest(familyPaths, regionName, assignSeqNum, null);
+ }
+
+ /**
+ * Create a protocol buffer bulk load request
+ *
+ * @param familyPaths
+ * @param regionName
+ * @param assignSeqNum
+ * @param clusterIds
+ * @return a bulk load request
+ */
+ public static BulkLoadHFileRequest buildBulkLoadHFileRequest(
final List<Pair<byte[], String>> familyPaths,
- final byte[] regionName, boolean assignSeqNum) {
+ final byte[] regionName, boolean assignSeqNum, List<String> clusterIds) {
BulkLoadHFileRequest.Builder builder = BulkLoadHFileRequest.newBuilder();
RegionSpecifier region = buildRegionSpecifier(
RegionSpecifierType.REGION_NAME, regionName);
@@ -566,6 +584,9 @@ public final class RequestConverter {
familyPathBuilder.setPath(familyPath.getSecond());
builder.addFamilyPath(familyPathBuilder.build());
}
+ if(clusterIds!=null) {
+ builder.addAllClusterIds(clusterIds);
+ }
builder.setAssignSeqNum(assignSeqNum);
return builder.build();
}
diff --git a/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/ClientProtos.java b/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/ClientProtos.java
index 611d1b5..b601bbc 100644
--- a/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/ClientProtos.java
+++ b/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/ClientProtos.java
@@ -1986,7 +1986,7 @@ public final class ClientProtos {
* <code>optional bool load_column_families_on_demand = 14;</code>
*
* <pre>
- * DO NOT add defaults to load_column_families_on_demand.
+ * DO NOT add defaults to load_column_families_on_demand.
* </pre>
*/
boolean hasLoadColumnFamiliesOnDemand();
@@ -1994,7 +1994,7 @@ public final class ClientProtos {
* <code>optional bool load_column_families_on_demand = 14;</code>
*
* <pre>
- * DO NOT add defaults to load_column_families_on_demand.
+ * DO NOT add defaults to load_column_families_on_demand.
* </pre>
*/
boolean getLoadColumnFamiliesOnDemand();
@@ -2515,7 +2515,7 @@ public final class ClientProtos {
* <code>optional bool load_column_families_on_demand = 14;</code>
*
* <pre>
- * DO NOT add defaults to load_column_families_on_demand.
+ * DO NOT add defaults to load_column_families_on_demand.
* </pre>
*/
public boolean hasLoadColumnFamiliesOnDemand() {
@@ -2525,7 +2525,7 @@ public final class ClientProtos {
* <code>optional bool load_column_families_on_demand = 14;</code>
*
* <pre>
- * DO NOT add defaults to load_column_families_on_demand.
+ * DO NOT add defaults to load_column_families_on_demand.
* </pre>
*/
public boolean getLoadColumnFamiliesOnDemand() {
@@ -4577,7 +4577,7 @@ public final class ClientProtos {
* <code>optional bool load_column_families_on_demand = 14;</code>
*
* <pre>
- * DO NOT add defaults to load_column_families_on_demand.
+ * DO NOT add defaults to load_column_families_on_demand.
* </pre>
*/
public boolean hasLoadColumnFamiliesOnDemand() {
@@ -4587,7 +4587,7 @@ public final class ClientProtos {
* <code>optional bool load_column_families_on_demand = 14;</code>
*
* <pre>
- * DO NOT add defaults to load_column_families_on_demand.
+ * DO NOT add defaults to load_column_families_on_demand.
* </pre>
*/
public boolean getLoadColumnFamiliesOnDemand() {
@@ -4597,7 +4597,7 @@ public final class ClientProtos {
* <code>optional bool load_column_families_on_demand = 14;</code>
*
* <pre>
- * DO NOT add defaults to load_column_families_on_demand.
+ * DO NOT add defaults to load_column_families_on_demand.
* </pre>
*/
public Builder setLoadColumnFamiliesOnDemand(boolean value) {
@@ -4610,7 +4610,7 @@ public final class ClientProtos {
* <code>optional bool load_column_families_on_demand = 14;</code>
*
* <pre>
- * DO NOT add defaults to load_column_families_on_demand.
+ * DO NOT add defaults to load_column_families_on_demand.
* </pre>
*/
public Builder clearLoadColumnFamiliesOnDemand() {
@@ -21302,6 +21302,26 @@ public final class ClientProtos {
* <code>optional bool assign_seq_num = 3;</code>
*/
boolean getAssignSeqNum();
+
+ // repeated string cluster_ids = 4;
+ /**
+ * <code>repeated string cluster_ids = 4;</code>
+ */
+ java.util.List<java.lang.String>
+ getClusterIdsList();
+ /**
+ * <code>repeated string cluster_ids = 4;</code>
+ */
+ int getClusterIdsCount();
+ /**
+ * <code>repeated string cluster_ids = 4;</code>
+ */
+ java.lang.String getClusterIds(int index);
+ /**
+ * <code>repeated string cluster_ids = 4;</code>
+ */
+ com.google.protobuf.ByteString
+ getClusterIdsBytes(int index);
}
/**
* Protobuf type {@code hbase.pb.BulkLoadHFileRequest}
@@ -21386,6 +21406,14 @@ public final class ClientProtos {
assignSeqNum_ = input.readBool();
break;
}
+ case 34: {
+ if (!((mutable_bitField0_ & 0x00000008) == 0x00000008)) {
+ clusterIds_ = new com.google.protobuf.LazyStringArrayList();
+ mutable_bitField0_ |= 0x00000008;
+ }
+ clusterIds_.add(input.readBytes());
+ break;
+ }
}
}
} catch (com.google.protobuf.InvalidProtocolBufferException e) {
@@ -21397,6 +21425,9 @@ public final class ClientProtos {
if (((mutable_bitField0_ & 0x00000002) == 0x00000002)) {
familyPath_ = java.util.Collections.unmodifiableList(familyPath_);
}
+ if (((mutable_bitField0_ & 0x00000008) == 0x00000008)) {
+ clusterIds_ = new com.google.protobuf.UnmodifiableLazyStringList(clusterIds_);
+ }
this.unknownFields = unknownFields.build();
makeExtensionsImmutable();
}
@@ -22121,10 +22152,41 @@ public final class ClientProtos {
return assignSeqNum_;
}
+ // repeated string cluster_ids = 4;
+ public static final int CLUSTER_IDS_FIELD_NUMBER = 4;
+ private com.google.protobuf.LazyStringList clusterIds_;
+ /**
+ * <code>repeated string cluster_ids = 4;</code>
+ */
+ public java.util.List<java.lang.String>
+ getClusterIdsList() {
+ return clusterIds_;
+ }
+ /**
+ * <code>repeated string cluster_ids = 4;</code>
+ */
+ public int getClusterIdsCount() {
+ return clusterIds_.size();
+ }
+ /**
+ * <code>repeated string cluster_ids = 4;</code>
+ */
+ public java.lang.String getClusterIds(int index) {
+ return clusterIds_.get(index);
+ }
+ /**
+ * <code>repeated string cluster_ids = 4;</code>
+ */
+ public com.google.protobuf.ByteString
+ getClusterIdsBytes(int index) {
+ return clusterIds_.getByteString(index);
+ }
+
private void initFields() {
region_ = org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.RegionSpecifier.getDefaultInstance();
familyPath_ = java.util.Collections.emptyList();
assignSeqNum_ = false;
+ clusterIds_ = com.google.protobuf.LazyStringArrayList.EMPTY;
}
private byte memoizedIsInitialized = -1;
public final boolean isInitialized() {
@@ -22161,6 +22223,9 @@ public final class ClientProtos {
if (((bitField0_ & 0x00000002) == 0x00000002)) {
output.writeBool(3, assignSeqNum_);
}
+ for (int i = 0; i < clusterIds_.size(); i++) {
+ output.writeBytes(4, clusterIds_.getByteString(i));
+ }
getUnknownFields().writeTo(output);
}
@@ -22182,6 +22247,15 @@ public final class ClientProtos {
size += com.google.protobuf.CodedOutputStream
.computeBoolSize(3, assignSeqNum_);
}
+ {
+ int dataSize = 0;
+ for (int i = 0; i < clusterIds_.size(); i++) {
+ dataSize += com.google.protobuf.CodedOutputStream
+ .computeBytesSizeNoTag(clusterIds_.getByteString(i));
+ }
+ size += dataSize;
+ size += 1 * getClusterIdsList().size();
+ }
size += getUnknownFields().getSerializedSize();
memoizedSerializedSize = size;
return size;
@@ -22217,6 +22291,8 @@ public final class ClientProtos {
result = result && (getAssignSeqNum()
== other.getAssignSeqNum());
}
+ result = result && getClusterIdsList()
+ .equals(other.getClusterIdsList());
result = result &&
getUnknownFields().equals(other.getUnknownFields());
return result;
@@ -22242,6 +22318,10 @@ public final class ClientProtos {
hash = (37 * hash) + ASSIGN_SEQ_NUM_FIELD_NUMBER;
hash = (53 * hash) + hashBoolean(getAssignSeqNum());
}
+ if (getClusterIdsCount() > 0) {
+ hash = (37 * hash) + CLUSTER_IDS_FIELD_NUMBER;
+ hash = (53 * hash) + getClusterIdsList().hashCode();
+ }
hash = (29 * hash) + getUnknownFields().hashCode();
memoizedHashCode = hash;
return hash;
@@ -22373,6 +22453,8 @@ public final class ClientProtos {
}
assignSeqNum_ = false;
bitField0_ = (bitField0_ & ~0x00000004);
+ clusterIds_ = com.google.protobuf.LazyStringArrayList.EMPTY;
+ bitField0_ = (bitField0_ & ~0x00000008);
return this;
}
@@ -22422,6 +22504,12 @@ public final class ClientProtos {
to_bitField0_ |= 0x00000002;
}
result.assignSeqNum_ = assignSeqNum_;
+ if (((bitField0_ & 0x00000008) == 0x00000008)) {
+ clusterIds_ = new com.google.protobuf.UnmodifiableLazyStringList(
+ clusterIds_);
+ bitField0_ = (bitField0_ & ~0x00000008);
+ }
+ result.clusterIds_ = clusterIds_;
result.bitField0_ = to_bitField0_;
onBuilt();
return result;
@@ -22470,6 +22558,16 @@ public final class ClientProtos {
if (other.hasAssignSeqNum()) {
setAssignSeqNum(other.getAssignSeqNum());
}
+ if (!other.clusterIds_.isEmpty()) {
+ if (clusterIds_.isEmpty()) {
+ clusterIds_ = other.clusterIds_;
+ bitField0_ = (bitField0_ & ~0x00000008);
+ } else {
+ ensureClusterIdsIsMutable();
+ clusterIds_.addAll(other.clusterIds_);
+ }
+ onChanged();
+ }
this.mergeUnknownFields(other.getUnknownFields());
return this;
}
@@ -22901,6 +22999,99 @@ public final class ClientProtos {
return this;
}
+ // repeated string cluster_ids = 4;
+ private com.google.protobuf.LazyStringList clusterIds_ = com.google.protobuf.LazyStringArrayList.EMPTY;
+ private void ensureClusterIdsIsMutable() {
+ if (!((bitField0_ & 0x00000008) == 0x00000008)) {
+ clusterIds_ = new com.google.protobuf.LazyStringArrayList(clusterIds_);
+ bitField0_ |= 0x00000008;
+ }
+ }
+ /**
+ * <code>repeated string cluster_ids = 4;</code>
+ */
+ public java.util.List<java.lang.String>
+ getClusterIdsList() {
+ return java.util.Collections.unmodifiableList(clusterIds_);
+ }
+ /**
+ * <code>repeated string cluster_ids = 4;</code>
+ */
+ public int getClusterIdsCount() {
+ return clusterIds_.size();
+ }
+ /**
+ * <code>repeated string cluster_ids = 4;</code>
+ */
+ public java.lang.String getClusterIds(int index) {
+ return clusterIds_.get(index);
+ }
+ /**
+ * <code>repeated string cluster_ids = 4;</code>
+ */
+ public com.google.protobuf.ByteString
+ getClusterIdsBytes(int index) {
+ return clusterIds_.getByteString(index);
+ }
+ /**
+ * <code>repeated string cluster_ids = 4;</code>
+ */
+ public Builder setClusterIds(
+ int index, java.lang.String value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ ensureClusterIdsIsMutable();
+ clusterIds_.set(index, value);
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>repeated string cluster_ids = 4;</code>
+ */
+ public Builder addClusterIds(
+ java.lang.String value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ ensureClusterIdsIsMutable();
+ clusterIds_.add(value);
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>repeated string cluster_ids = 4;</code>
+ */
+ public Builder addAllClusterIds(
+ java.lang.Iterable<java.lang.String> values) {
+ ensureClusterIdsIsMutable();
+ super.addAll(values, clusterIds_);
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>repeated string cluster_ids = 4;</code>
+ */
+ public Builder clearClusterIds() {
+ clusterIds_ = com.google.protobuf.LazyStringArrayList.EMPTY;
+ bitField0_ = (bitField0_ & ~0x00000008);
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>repeated string cluster_ids = 4;</code>
+ */
+ public Builder addClusterIdsBytes(
+ com.google.protobuf.ByteString value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ ensureClusterIdsIsMutable();
+ clusterIds_.add(value);
+ onChanged();
+ return this;
+ }
+
// @@protoc_insertion_point(builder_scope:hbase.pb.BulkLoadHFileRequest)
}
@@ -35823,66 +36014,67 @@ public final class ClientProtos {
"sult\022\r\n\005stale\030\006 \001(\010\022\037\n\027partial_flag_per_" +
"result\030\007 \003(\010\022\036\n\026more_results_in_region\030\010" +
" \001(\010\022\031\n\021heartbeat_message\030\t \001(\010\022+\n\014scan_" +
- "metrics\030\n \001(\0132\025.hbase.pb.ScanMetrics\"\305\001\n" +
+ "metrics\030\n \001(\0132\025.hbase.pb.ScanMetrics\"\332\001\n" +
"\024BulkLoadHFileRequest\022)\n\006region\030\001 \002(\0132\031." +
"hbase.pb.RegionSpecifier\022>\n\013family_path\030" +
"\002 \003(\0132).hbase.pb.BulkLoadHFileRequest.Fa" +
- "milyPath\022\026\n\016assign_seq_num\030\003 \001(\010\032*\n\nFami" +
- "lyPath\022\016\n\006family\030\001 \002(\014\022\014\n\004path\030\002 \002(\t\"\'\n\025",
- "BulkLoadHFileResponse\022\016\n\006loaded\030\001 \002(\010\"a\n" +
- "\026CoprocessorServiceCall\022\013\n\003row\030\001 \002(\014\022\024\n\014" +
- "service_name\030\002 \002(\t\022\023\n\013method_name\030\003 \002(\t\022" +
- "\017\n\007request\030\004 \002(\014\"B\n\030CoprocessorServiceRe" +
- "sult\022&\n\005value\030\001 \001(\0132\027.hbase.pb.NameBytes" +
- "Pair\"v\n\031CoprocessorServiceRequest\022)\n\006reg" +
- "ion\030\001 \002(\0132\031.hbase.pb.RegionSpecifier\022.\n\004" +
- "call\030\002 \002(\0132 .hbase.pb.CoprocessorService" +
- "Call\"o\n\032CoprocessorServiceResponse\022)\n\006re" +
- "gion\030\001 \002(\0132\031.hbase.pb.RegionSpecifier\022&\n",
- "\005value\030\002 \002(\0132\027.hbase.pb.NameBytesPair\"\226\001" +
- "\n\006Action\022\r\n\005index\030\001 \001(\r\022)\n\010mutation\030\002 \001(" +
- "\0132\027.hbase.pb.MutationProto\022\032\n\003get\030\003 \001(\0132" +
- "\r.hbase.pb.Get\0226\n\014service_call\030\004 \001(\0132 .h" +
- "base.pb.CoprocessorServiceCall\"k\n\014Region" +
- "Action\022)\n\006region\030\001 \002(\0132\031.hbase.pb.Region" +
- "Specifier\022\016\n\006atomic\030\002 \001(\010\022 \n\006action\030\003 \003(" +
- "\0132\020.hbase.pb.Action\"c\n\017RegionLoadStats\022\027" +
- "\n\014memstoreLoad\030\001 \001(\005:\0010\022\030\n\rheapOccupancy" +
- "\030\002 \001(\005:\0010\022\035\n\022compactionPressure\030\003 \001(\005:\0010",
- "\"j\n\024MultiRegionLoadStats\022)\n\006region\030\001 \003(\013" +
- "2\031.hbase.pb.RegionSpecifier\022\'\n\004stat\030\002 \003(" +
- "\0132\031.hbase.pb.RegionLoadStats\"\336\001\n\021ResultO" +
- "rException\022\r\n\005index\030\001 \001(\r\022 \n\006result\030\002 \001(" +
- "\0132\020.hbase.pb.Result\022*\n\texception\030\003 \001(\0132\027" +
- ".hbase.pb.NameBytesPair\022:\n\016service_resul" +
- "t\030\004 \001(\0132\".hbase.pb.CoprocessorServiceRes" +
- "ult\0220\n\tloadStats\030\005 \001(\0132\031.hbase.pb.Region" +
- "LoadStatsB\002\030\001\"x\n\022RegionActionResult\0226\n\021r" +
- "esultOrException\030\001 \003(\0132\033.hbase.pb.Result",
- "OrException\022*\n\texception\030\002 \001(\0132\027.hbase.p" +
- "b.NameBytesPair\"x\n\014MultiRequest\022,\n\014regio" +
- "nAction\030\001 \003(\0132\026.hbase.pb.RegionAction\022\022\n" +
- "\nnonceGroup\030\002 \001(\004\022&\n\tcondition\030\003 \001(\0132\023.h" +
- "base.pb.Condition\"\226\001\n\rMultiResponse\0228\n\022r" +
- "egionActionResult\030\001 \003(\0132\034.hbase.pb.Regio" +
- "nActionResult\022\021\n\tprocessed\030\002 \001(\010\0228\n\020regi" +
- "onStatistics\030\003 \001(\0132\036.hbase.pb.MultiRegio" +
- "nLoadStats*\'\n\013Consistency\022\n\n\006STRONG\020\000\022\014\n" +
- "\010TIMELINE\020\0012\203\004\n\rClientService\0222\n\003Get\022\024.h",
- "base.pb.GetRequest\032\025.hbase.pb.GetRespons" +
- "e\022;\n\006Mutate\022\027.hbase.pb.MutateRequest\032\030.h" +
- "base.pb.MutateResponse\0225\n\004Scan\022\025.hbase.p" +
- "b.ScanRequest\032\026.hbase.pb.ScanResponse\022P\n" +
- "\rBulkLoadHFile\022\036.hbase.pb.BulkLoadHFileR" +
- "equest\032\037.hbase.pb.BulkLoadHFileResponse\022" +
- "X\n\013ExecService\022#.hbase.pb.CoprocessorSer" +
- "viceRequest\032$.hbase.pb.CoprocessorServic" +
- "eResponse\022d\n\027ExecRegionServerService\022#.h" +
- "base.pb.CoprocessorServiceRequest\032$.hbas",
- "e.pb.CoprocessorServiceResponse\0228\n\005Multi" +
- "\022\026.hbase.pb.MultiRequest\032\027.hbase.pb.Mult" +
- "iResponseBB\n*org.apache.hadoop.hbase.pro" +
- "tobuf.generatedB\014ClientProtosH\001\210\001\001\240\001\001"
+ "milyPath\022\026\n\016assign_seq_num\030\003 \001(\010\022\023\n\013clus" +
+ "ter_ids\030\004 \003(\t\032*\n\nFamilyPath\022\016\n\006family\030\001 ",
+ "\002(\014\022\014\n\004path\030\002 \002(\t\"\'\n\025BulkLoadHFileRespon" +
+ "se\022\016\n\006loaded\030\001 \002(\010\"a\n\026CoprocessorService" +
+ "Call\022\013\n\003row\030\001 \002(\014\022\024\n\014service_name\030\002 \002(\t\022" +
+ "\023\n\013method_name\030\003 \002(\t\022\017\n\007request\030\004 \002(\014\"B\n" +
+ "\030CoprocessorServiceResult\022&\n\005value\030\001 \001(\013" +
+ "2\027.hbase.pb.NameBytesPair\"v\n\031Coprocessor" +
+ "ServiceRequest\022)\n\006region\030\001 \002(\0132\031.hbase.p" +
+ "b.RegionSpecifier\022.\n\004call\030\002 \002(\0132 .hbase." +
+ "pb.CoprocessorServiceCall\"o\n\032Coprocessor" +
+ "ServiceResponse\022)\n\006region\030\001 \002(\0132\031.hbase.",
+ "pb.RegionSpecifier\022&\n\005value\030\002 \002(\0132\027.hbas" +
+ "e.pb.NameBytesPair\"\226\001\n\006Action\022\r\n\005index\030\001" +
+ " \001(\r\022)\n\010mutation\030\002 \001(\0132\027.hbase.pb.Mutati" +
+ "onProto\022\032\n\003get\030\003 \001(\0132\r.hbase.pb.Get\0226\n\014s" +
+ "ervice_call\030\004 \001(\0132 .hbase.pb.Coprocessor" +
+ "ServiceCall\"k\n\014RegionAction\022)\n\006region\030\001 " +
+ "\002(\0132\031.hbase.pb.RegionSpecifier\022\016\n\006atomic" +
+ "\030\002 \001(\010\022 \n\006action\030\003 \003(\0132\020.hbase.pb.Action" +
+ "\"c\n\017RegionLoadStats\022\027\n\014memstoreLoad\030\001 \001(" +
+ "\005:\0010\022\030\n\rheapOccupancy\030\002 \001(\005:\0010\022\035\n\022compac",
+ "tionPressure\030\003 \001(\005:\0010\"j\n\024MultiRegionLoad" +
+ "Stats\022)\n\006region\030\001 \003(\0132\031.hbase.pb.RegionS" +
+ "pecifier\022\'\n\004stat\030\002 \003(\0132\031.hbase.pb.Region" +
+ "LoadStats\"\336\001\n\021ResultOrException\022\r\n\005index" +
+ "\030\001 \001(\r\022 \n\006result\030\002 \001(\0132\020.hbase.pb.Result" +
+ "\022*\n\texception\030\003 \001(\0132\027.hbase.pb.NameBytes" +
+ "Pair\022:\n\016service_result\030\004 \001(\0132\".hbase.pb." +
+ "CoprocessorServiceResult\0220\n\tloadStats\030\005 " +
+ "\001(\0132\031.hbase.pb.RegionLoadStatsB\002\030\001\"x\n\022Re" +
+ "gionActionResult\0226\n\021resultOrException\030\001 ",
+ "\003(\0132\033.hbase.pb.ResultOrException\022*\n\texce" +
+ "ption\030\002 \001(\0132\027.hbase.pb.NameBytesPair\"x\n\014" +
+ "MultiRequest\022,\n\014regionAction\030\001 \003(\0132\026.hba" +
+ "se.pb.RegionAction\022\022\n\nnonceGroup\030\002 \001(\004\022&" +
+ "\n\tcondition\030\003 \001(\0132\023.hbase.pb.Condition\"\226" +
+ "\001\n\rMultiResponse\0228\n\022regionActionResult\030\001" +
+ " \003(\0132\034.hbase.pb.RegionActionResult\022\021\n\tpr" +
+ "ocessed\030\002 \001(\010\0228\n\020regionStatistics\030\003 \001(\0132" +
+ "\036.hbase.pb.MultiRegionLoadStats*\'\n\013Consi" +
+ "stency\022\n\n\006STRONG\020\000\022\014\n\010TIMELINE\020\0012\203\004\n\rCli",
+ "entService\0222\n\003Get\022\024.hbase.pb.GetRequest\032" +
+ "\025.hbase.pb.GetResponse\022;\n\006Mutate\022\027.hbase" +
+ ".pb.MutateRequest\032\030.hbase.pb.MutateRespo" +
+ "nse\0225\n\004Scan\022\025.hbase.pb.ScanRequest\032\026.hba" +
+ "se.pb.ScanResponse\022P\n\rBulkLoadHFile\022\036.hb" +
+ "ase.pb.BulkLoadHFileRequest\032\037.hbase.pb.B" +
+ "ulkLoadHFileResponse\022X\n\013ExecService\022#.hb" +
+ "ase.pb.CoprocessorServiceRequest\032$.hbase" +
+ ".pb.CoprocessorServiceResponse\022d\n\027ExecRe" +
+ "gionServerService\022#.hbase.pb.Coprocessor",
+ "ServiceRequest\032$.hbase.pb.CoprocessorSer" +
+ "viceResponse\0228\n\005Multi\022\026.hbase.pb.MultiRe" +
+ "quest\032\027.hbase.pb.MultiResponseBB\n*org.ap" +
+ "ache.hadoop.hbase.protobuf.generatedB\014Cl" +
+ "ientProtosH\001\210\001\001\240\001\001"
};
com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
@@ -35990,7 +36182,7 @@ public final class ClientProtos {
internal_static_hbase_pb_BulkLoadHFileRequest_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_hbase_pb_BulkLoadHFileRequest_descriptor,
- new java.lang.String[] { "Region", "FamilyPath", "AssignSeqNum", });
+ new java.lang.String[] { "Region", "FamilyPath", "AssignSeqNum", "ClusterIds", });
internal_static_hbase_pb_BulkLoadHFileRequest_FamilyPath_descriptor =
internal_static_hbase_pb_BulkLoadHFileRequest_descriptor.getNestedTypes().get(0);
internal_static_hbase_pb_BulkLoadHFileRequest_FamilyPath_fieldAccessorTable = new
diff --git a/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/SecureBulkLoadProtos.java b/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/SecureBulkLoadProtos.java
index 538e031..a1fa93d 100644
--- a/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/SecureBulkLoadProtos.java
+++ b/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/SecureBulkLoadProtos.java
@@ -74,6 +74,26 @@ public final class SecureBulkLoadProtos {
*/
com.google.protobuf.ByteString
getBulkTokenBytes();
+
+ // repeated string cluster_ids = 5;
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ java.util.List<java.lang.String>
+ getClusterIdsList();
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ int getClusterIdsCount();
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ java.lang.String getClusterIds(int index);
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ com.google.protobuf.ByteString
+ getClusterIdsBytes(int index);
}
/**
* Protobuf type {@code hbase.pb.SecureBulkLoadHFilesRequest}
@@ -157,6 +177,14 @@ public final class SecureBulkLoadProtos {
bulkToken_ = input.readBytes();
break;
}
+ case 42: {
+ if (!((mutable_bitField0_ & 0x00000010) == 0x00000010)) {
+ clusterIds_ = new com.google.protobuf.LazyStringArrayList();
+ mutable_bitField0_ |= 0x00000010;
+ }
+ clusterIds_.add(input.readBytes());
+ break;
+ }
}
}
} catch (com.google.protobuf.InvalidProtocolBufferException e) {
@@ -168,6 +196,9 @@ public final class SecureBulkLoadProtos {
if (((mutable_bitField0_ & 0x00000001) == 0x00000001)) {
familyPath_ = java.util.Collections.unmodifiableList(familyPath_);
}
+ if (((mutable_bitField0_ & 0x00000010) == 0x00000010)) {
+ clusterIds_ = new com.google.protobuf.UnmodifiableLazyStringList(clusterIds_);
+ }
this.unknownFields = unknownFields.build();
makeExtensionsImmutable();
}
@@ -317,11 +348,42 @@ public final class SecureBulkLoadProtos {
}
}
+ // repeated string cluster_ids = 5;
+ public static final int CLUSTER_IDS_FIELD_NUMBER = 5;
+ private com.google.protobuf.LazyStringList clusterIds_;
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public java.util.List<java.lang.String>
+ getClusterIdsList() {
+ return clusterIds_;
+ }
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public int getClusterIdsCount() {
+ return clusterIds_.size();
+ }
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public java.lang.String getClusterIds(int index) {
+ return clusterIds_.get(index);
+ }
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public com.google.protobuf.ByteString
+ getClusterIdsBytes(int index) {
+ return clusterIds_.getByteString(index);
+ }
+
private void initFields() {
familyPath_ = java.util.Collections.emptyList();
assignSeqNum_ = false;
fsToken_ = org.apache.hadoop.hbase.protobuf.generated.SecureBulkLoadProtos.DelegationToken.getDefaultInstance();
bulkToken_ = "";
+ clusterIds_ = com.google.protobuf.LazyStringArrayList.EMPTY;
}
private byte memoizedIsInitialized = -1;
public final boolean isInitialized() {
@@ -361,6 +423,9 @@ public final class SecureBulkLoadProtos {
if (((bitField0_ & 0x00000004) == 0x00000004)) {
output.writeBytes(4, getBulkTokenBytes());
}
+ for (int i = 0; i < clusterIds_.size(); i++) {
+ output.writeBytes(5, clusterIds_.getByteString(i));
+ }
getUnknownFields().writeTo(output);
}
@@ -386,6 +451,15 @@ public final class SecureBulkLoadProtos {
size += com.google.protobuf.CodedOutputStream
.computeBytesSize(4, getBulkTokenBytes());
}
+ {
+ int dataSize = 0;
+ for (int i = 0; i < clusterIds_.size(); i++) {
+ dataSize += com.google.protobuf.CodedOutputStream
+ .computeBytesSizeNoTag(clusterIds_.getByteString(i));
+ }
+ size += dataSize;
+ size += 1 * getClusterIdsList().size();
+ }
size += getUnknownFields().getSerializedSize();
memoizedSerializedSize = size;
return size;
@@ -426,6 +500,8 @@ public final class SecureBulkLoadProtos {
result = result && getBulkToken()
.equals(other.getBulkToken());
}
+ result = result && getClusterIdsList()
+ .equals(other.getClusterIdsList());
result = result &&
getUnknownFields().equals(other.getUnknownFields());
return result;
@@ -455,6 +531,10 @@ public final class SecureBulkLoadProtos {
hash = (37 * hash) + BULK_TOKEN_FIELD_NUMBER;
hash = (53 * hash) + getBulkToken().hashCode();
}
+ if (getClusterIdsCount() > 0) {
+ hash = (37 * hash) + CLUSTER_IDS_FIELD_NUMBER;
+ hash = (53 * hash) + getClusterIdsList().hashCode();
+ }
hash = (29 * hash) + getUnknownFields().hashCode();
memoizedHashCode = hash;
return hash;
@@ -582,6 +662,8 @@ public final class SecureBulkLoadProtos {
bitField0_ = (bitField0_ & ~0x00000004);
bulkToken_ = "";
bitField0_ = (bitField0_ & ~0x00000008);
+ clusterIds_ = com.google.protobuf.LazyStringArrayList.EMPTY;
+ bitField0_ = (bitField0_ & ~0x00000010);
return this;
}
@@ -635,6 +717,12 @@ public final class SecureBulkLoadProtos {
to_bitField0_ |= 0x00000004;
}
result.bulkToken_ = bulkToken_;
+ if (((bitField0_ & 0x00000010) == 0x00000010)) {
+ clusterIds_ = new com.google.protobuf.UnmodifiableLazyStringList(
+ clusterIds_);
+ bitField0_ = (bitField0_ & ~0x00000010);
+ }
+ result.clusterIds_ = clusterIds_;
result.bitField0_ = to_bitField0_;
onBuilt();
return result;
@@ -688,6 +776,16 @@ public final class SecureBulkLoadProtos {
bulkToken_ = other.bulkToken_;
onChanged();
}
+ if (!other.clusterIds_.isEmpty()) {
+ if (clusterIds_.isEmpty()) {
+ clusterIds_ = other.clusterIds_;
+ bitField0_ = (bitField0_ & ~0x00000010);
+ } else {
+ ensureClusterIdsIsMutable();
+ clusterIds_.addAll(other.clusterIds_);
+ }
+ onChanged();
+ }
this.mergeUnknownFields(other.getUnknownFields());
return this;
}
@@ -1193,6 +1291,99 @@ public final class SecureBulkLoadProtos {
return this;
}
+ // repeated string cluster_ids = 5;
+ private com.google.protobuf.LazyStringList clusterIds_ = com.google.protobuf.LazyStringArrayList.EMPTY;
+ private void ensureClusterIdsIsMutable() {
+ if (!((bitField0_ & 0x00000010) == 0x00000010)) {
+ clusterIds_ = new com.google.protobuf.LazyStringArrayList(clusterIds_);
+ bitField0_ |= 0x00000010;
+ }
+ }
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public java.util.List<java.lang.String>
+ getClusterIdsList() {
+ return java.util.Collections.unmodifiableList(clusterIds_);
+ }
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public int getClusterIdsCount() {
+ return clusterIds_.size();
+ }
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public java.lang.String getClusterIds(int index) {
+ return clusterIds_.get(index);
+ }
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public com.google.protobuf.ByteString
+ getClusterIdsBytes(int index) {
+ return clusterIds_.getByteString(index);
+ }
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public Builder setClusterIds(
+ int index, java.lang.String value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ ensureClusterIdsIsMutable();
+ clusterIds_.set(index, value);
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public Builder addClusterIds(
+ java.lang.String value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ ensureClusterIdsIsMutable();
+ clusterIds_.add(value);
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public Builder addAllClusterIds(
+ java.lang.Iterable<java.lang.String> values) {
+ ensureClusterIdsIsMutable();
+ super.addAll(values, clusterIds_);
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public Builder clearClusterIds() {
+ clusterIds_ = com.google.protobuf.LazyStringArrayList.EMPTY;
+ bitField0_ = (bitField0_ & ~0x00000010);
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public Builder addClusterIdsBytes(
+ com.google.protobuf.ByteString value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ ensureClusterIdsIsMutable();
+ clusterIds_.add(value);
+ onChanged();
+ return this;
+ }
+
// @@protoc_insertion_point(builder_scope:hbase.pb.SecureBulkLoadHFilesRequest)
}
@@ -4858,29 +5049,30 @@ public final class SecureBulkLoadProtos {
static {
java.lang.String[] descriptorData = {
"\n\024SecureBulkLoad.proto\022\010hbase.pb\032\013HBase." +
- "proto\032\014Client.proto\"\266\001\n\033SecureBulkLoadHF" +
+ "proto\032\014Client.proto\"\313\001\n\033SecureBulkLoadHF" +
"ilesRequest\022>\n\013family_path\030\001 \003(\0132).hbase" +
".pb.BulkLoadHFileRequest.FamilyPath\022\026\n\016a" +
"ssign_seq_num\030\002 \001(\010\022+\n\010fs_token\030\003 \002(\0132\031." +
"hbase.pb.DelegationToken\022\022\n\nbulk_token\030\004" +
- " \002(\t\".\n\034SecureBulkLoadHFilesResponse\022\016\n\006" +
- "loaded\030\001 \002(\010\"V\n\017DelegationToken\022\022\n\nident" +
- "ifier\030\001 \001(\014\022\020\n\010password\030\002 \001(\014\022\014\n\004kind\030\003 " +
- "\001(\t\022\017\n\007service\030\004 \001(\t\"A\n\026PrepareBulkLoadR",
- "equest\022\'\n\ntable_name\030\001 \002(\0132\023.hbase.pb.Ta" +
- "bleName\"-\n\027PrepareBulkLoadResponse\022\022\n\nbu" +
- "lk_token\030\001 \002(\t\",\n\026CleanupBulkLoadRequest" +
- "\022\022\n\nbulk_token\030\001 \002(\t\"\031\n\027CleanupBulkLoadR" +
- "esponse2\256\002\n\025SecureBulkLoadService\022V\n\017Pre" +
- "pareBulkLoad\022 .hbase.pb.PrepareBulkLoadR" +
- "equest\032!.hbase.pb.PrepareBulkLoadRespons" +
- "e\022e\n\024SecureBulkLoadHFiles\022%.hbase.pb.Sec" +
- "ureBulkLoadHFilesRequest\032&.hbase.pb.Secu" +
- "reBulkLoadHFilesResponse\022V\n\017CleanupBulkL",
- "oad\022 .hbase.pb.CleanupBulkLoadRequest\032!." +
- "hbase.pb.CleanupBulkLoadResponseBJ\n*org." +
- "apache.hadoop.hbase.protobuf.generatedB\024" +
- "SecureBulkLoadProtosH\001\210\001\001\240\001\001"
+ " \002(\t\022\023\n\013cluster_ids\030\005 \003(\t\".\n\034SecureBulkL" +
+ "oadHFilesResponse\022\016\n\006loaded\030\001 \002(\010\"V\n\017Del" +
+ "egationToken\022\022\n\nidentifier\030\001 \001(\014\022\020\n\010pass" +
+ "word\030\002 \001(\014\022\014\n\004kind\030\003 \001(\t\022\017\n\007service\030\004 \001(",
+ "\t\"A\n\026PrepareBulkLoadRequest\022\'\n\ntable_nam" +
+ "e\030\001 \002(\0132\023.hbase.pb.TableName\"-\n\027PrepareB" +
+ "ulkLoadResponse\022\022\n\nbulk_token\030\001 \002(\t\",\n\026C" +
+ "leanupBulkLoadRequest\022\022\n\nbulk_token\030\001 \002(" +
+ "\t\"\031\n\027CleanupBulkLoadResponse2\256\002\n\025SecureB" +
+ "ulkLoadService\022V\n\017PrepareBulkLoad\022 .hbas" +
+ "e.pb.PrepareBulkLoadRequest\032!.hbase.pb.P" +
+ "repareBulkLoadResponse\022e\n\024SecureBulkLoad" +
+ "HFiles\022%.hbase.pb.SecureBulkLoadHFilesRe" +
+ "quest\032&.hbase.pb.SecureBulkLoadHFilesRes",
+ "ponse\022V\n\017CleanupBulkLoad\022 .hbase.pb.Clea" +
+ "nupBulkLoadRequest\032!.hbase.pb.CleanupBul" +
+ "kLoadResponseBJ\n*org.apache.hadoop.hbase" +
+ ".protobuf.generatedB\024SecureBulkLoadProto" +
+ "sH\001\210\001\001\240\001\001"
};
com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
@@ -4892,7 +5084,7 @@ public final class SecureBulkLoadProtos {
internal_static_hbase_pb_SecureBulkLoadHFilesRequest_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_hbase_pb_SecureBulkLoadHFilesRequest_descriptor,
- new java.lang.String[] { "FamilyPath", "AssignSeqNum", "FsToken", "BulkToken", });
+ new java.lang.String[] { "FamilyPath", "AssignSeqNum", "FsToken", "BulkToken", "ClusterIds", });
internal_static_hbase_pb_SecureBulkLoadHFilesResponse_descriptor =
getDescriptor().getMessageTypes().get(1);
internal_static_hbase_pb_SecureBulkLoadHFilesResponse_fieldAccessorTable = new
diff --git a/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/WALProtos.java b/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/WALProtos.java
index 4da6fa2..6678334 100644
--- a/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/WALProtos.java
+++ b/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/WALProtos.java
@@ -8854,6 +8854,26 @@ public final class WALProtos {
* <code>required int64 bulkload_seq_num = 4;</code>
*/
long getBulkloadSeqNum();
+
+ // repeated string cluster_ids = 5;
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ java.util.List<java.lang.String>
+ getClusterIdsList();
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ int getClusterIdsCount();
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ java.lang.String getClusterIds(int index);
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ com.google.protobuf.ByteString
+ getClusterIdsBytes(int index);
}
/**
* Protobuf type {@code hbase.pb.BulkLoadDescriptor}
@@ -8942,6 +8962,14 @@ public final class WALProtos {
bulkloadSeqNum_ = input.readInt64();
break;
}
+ case 42: {
+ if (!((mutable_bitField0_ & 0x00000010) == 0x00000010)) {
+ clusterIds_ = new com.google.protobuf.LazyStringArrayList();
+ mutable_bitField0_ |= 0x00000010;
+ }
+ clusterIds_.add(input.readBytes());
+ break;
+ }
}
}
} catch (com.google.protobuf.InvalidProtocolBufferException e) {
@@ -8953,6 +8981,9 @@ public final class WALProtos {
if (((mutable_bitField0_ & 0x00000004) == 0x00000004)) {
stores_ = java.util.Collections.unmodifiableList(stores_);
}
+ if (((mutable_bitField0_ & 0x00000010) == 0x00000010)) {
+ clusterIds_ = new com.google.protobuf.UnmodifiableLazyStringList(clusterIds_);
+ }
this.unknownFields = unknownFields.build();
makeExtensionsImmutable();
}
@@ -9075,11 +9106,42 @@ public final class WALProtos {
return bulkloadSeqNum_;
}
+ // repeated string cluster_ids = 5;
+ public static final int CLUSTER_IDS_FIELD_NUMBER = 5;
+ private com.google.protobuf.LazyStringList clusterIds_;
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public java.util.List<java.lang.String>
+ getClusterIdsList() {
+ return clusterIds_;
+ }
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public int getClusterIdsCount() {
+ return clusterIds_.size();
+ }
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public java.lang.String getClusterIds(int index) {
+ return clusterIds_.get(index);
+ }
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public com.google.protobuf.ByteString
+ getClusterIdsBytes(int index) {
+ return clusterIds_.getByteString(index);
+ }
+
private void initFields() {
tableName_ = org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.TableName.getDefaultInstance();
encodedRegionName_ = com.google.protobuf.ByteString.EMPTY;
stores_ = java.util.Collections.emptyList();
bulkloadSeqNum_ = 0L;
+ clusterIds_ = com.google.protobuf.LazyStringArrayList.EMPTY;
}
private byte memoizedIsInitialized = -1;
public final boolean isInitialized() {
@@ -9127,6 +9189,9 @@ public final class WALProtos {
if (((bitField0_ & 0x00000004) == 0x00000004)) {
output.writeInt64(4, bulkloadSeqNum_);
}
+ for (int i = 0; i < clusterIds_.size(); i++) {
+ output.writeBytes(5, clusterIds_.getByteString(i));
+ }
getUnknownFields().writeTo(output);
}
@@ -9152,6 +9217,15 @@ public final class WALProtos {
size += com.google.protobuf.CodedOutputStream
.computeInt64Size(4, bulkloadSeqNum_);
}
+ {
+ int dataSize = 0;
+ for (int i = 0; i < clusterIds_.size(); i++) {
+ dataSize += com.google.protobuf.CodedOutputStream
+ .computeBytesSizeNoTag(clusterIds_.getByteString(i));
+ }
+ size += dataSize;
+ size += 1 * getClusterIdsList().size();
+ }
size += getUnknownFields().getSerializedSize();
memoizedSerializedSize = size;
return size;
@@ -9192,6 +9266,8 @@ public final class WALProtos {
result = result && (getBulkloadSeqNum()
== other.getBulkloadSeqNum());
}
+ result = result && getClusterIdsList()
+ .equals(other.getClusterIdsList());
result = result &&
getUnknownFields().equals(other.getUnknownFields());
return result;
@@ -9221,6 +9297,10 @@ public final class WALProtos {
hash = (37 * hash) + BULKLOAD_SEQ_NUM_FIELD_NUMBER;
hash = (53 * hash) + hashLong(getBulkloadSeqNum());
}
+ if (getClusterIdsCount() > 0) {
+ hash = (37 * hash) + CLUSTER_IDS_FIELD_NUMBER;
+ hash = (53 * hash) + getClusterIdsList().hashCode();
+ }
hash = (29 * hash) + getUnknownFields().hashCode();
memoizedHashCode = hash;
return hash;
@@ -9353,6 +9433,8 @@ public final class WALProtos {
}
bulkloadSeqNum_ = 0L;
bitField0_ = (bitField0_ & ~0x00000008);
+ clusterIds_ = com.google.protobuf.LazyStringArrayList.EMPTY;
+ bitField0_ = (bitField0_ & ~0x00000010);
return this;
}
@@ -9406,6 +9488,12 @@ public final class WALProtos {
to_bitField0_ |= 0x00000004;
}
result.bulkloadSeqNum_ = bulkloadSeqNum_;
+ if (((bitField0_ & 0x00000010) == 0x00000010)) {
+ clusterIds_ = new com.google.protobuf.UnmodifiableLazyStringList(
+ clusterIds_);
+ bitField0_ = (bitField0_ & ~0x00000010);
+ }
+ result.clusterIds_ = clusterIds_;
result.bitField0_ = to_bitField0_;
onBuilt();
return result;
@@ -9457,6 +9545,16 @@ public final class WALProtos {
if (other.hasBulkloadSeqNum()) {
setBulkloadSeqNum(other.getBulkloadSeqNum());
}
+ if (!other.clusterIds_.isEmpty()) {
+ if (clusterIds_.isEmpty()) {
+ clusterIds_ = other.clusterIds_;
+ bitField0_ = (bitField0_ & ~0x00000010);
+ } else {
+ ensureClusterIdsIsMutable();
+ clusterIds_.addAll(other.clusterIds_);
+ }
+ onChanged();
+ }
this.mergeUnknownFields(other.getUnknownFields());
return this;
}
@@ -9932,6 +10030,99 @@ public final class WALProtos {
return this;
}
+ // repeated string cluster_ids = 5;
+ private com.google.protobuf.LazyStringList clusterIds_ = com.google.protobuf.LazyStringArrayList.EMPTY;
+ private void ensureClusterIdsIsMutable() {
+ if (!((bitField0_ & 0x00000010) == 0x00000010)) {
+ clusterIds_ = new com.google.protobuf.LazyStringArrayList(clusterIds_);
+ bitField0_ |= 0x00000010;
+ }
+ }
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public java.util.List<java.lang.String>
+ getClusterIdsList() {
+ return java.util.Collections.unmodifiableList(clusterIds_);
+ }
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public int getClusterIdsCount() {
+ return clusterIds_.size();
+ }
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public java.lang.String getClusterIds(int index) {
+ return clusterIds_.get(index);
+ }
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public com.google.protobuf.ByteString
+ getClusterIdsBytes(int index) {
+ return clusterIds_.getByteString(index);
+ }
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public Builder setClusterIds(
+ int index, java.lang.String value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ ensureClusterIdsIsMutable();
+ clusterIds_.set(index, value);
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public Builder addClusterIds(
+ java.lang.String value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ ensureClusterIdsIsMutable();
+ clusterIds_.add(value);
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public Builder addAllClusterIds(
+ java.lang.Iterable<java.lang.String> values) {
+ ensureClusterIdsIsMutable();
+ super.addAll(values, clusterIds_);
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public Builder clearClusterIds() {
+ clusterIds_ = com.google.protobuf.LazyStringArrayList.EMPTY;
+ bitField0_ = (bitField0_ & ~0x00000010);
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>repeated string cluster_ids = 5;</code>
+ */
+ public Builder addClusterIdsBytes(
+ com.google.protobuf.ByteString value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ ensureClusterIdsIsMutable();
+ clusterIds_.add(value);
+ onChanged();
+ return this;
+ }
+
// @@protoc_insertion_point(builder_scope:hbase.pb.BulkLoadDescriptor)
}
@@ -12001,23 +12192,24 @@ public final class WALProtos {
"USH\020\001\022\017\n\013ABORT_FLUSH\020\002\022\020\n\014CANNOT_FLUSH\020\003",
"\"q\n\017StoreDescriptor\022\023\n\013family_name\030\001 \002(\014" +
"\022\026\n\016store_home_dir\030\002 \002(\t\022\022\n\nstore_file\030\003" +
- " \003(\t\022\035\n\025store_file_size_bytes\030\004 \001(\004\"\237\001\n\022" +
+ " \003(\t\022\035\n\025store_file_size_bytes\030\004 \001(\004\"\264\001\n\022" +
"BulkLoadDescriptor\022\'\n\ntable_name\030\001 \002(\0132\023" +
".hbase.pb.TableName\022\033\n\023encoded_region_na" +
"me\030\002 \002(\014\022)\n\006stores\030\003 \003(\0132\031.hbase.pb.Stor" +
- "eDescriptor\022\030\n\020bulkload_seq_num\030\004 \002(\003\"\272\002" +
- "\n\025RegionEventDescriptor\022=\n\nevent_type\030\001 " +
- "\002(\0162).hbase.pb.RegionEventDescriptor.Eve" +
- "ntType\022\022\n\ntable_name\030\002 \002(\014\022\033\n\023encoded_re",
- "gion_name\030\003 \002(\014\022\033\n\023log_sequence_number\030\004" +
- " \001(\004\022)\n\006stores\030\005 \003(\0132\031.hbase.pb.StoreDes" +
- "criptor\022$\n\006server\030\006 \001(\0132\024.hbase.pb.Serve" +
- "rName\022\023\n\013region_name\030\007 \001(\014\".\n\tEventType\022" +
- "\017\n\013REGION_OPEN\020\000\022\020\n\014REGION_CLOSE\020\001\"\014\n\nWA" +
- "LTrailer*F\n\tScopeType\022\033\n\027REPLICATION_SCO" +
- "PE_LOCAL\020\000\022\034\n\030REPLICATION_SCOPE_GLOBAL\020\001" +
- "B?\n*org.apache.hadoop.hbase.protobuf.gen" +
- "eratedB\tWALProtosH\001\210\001\000\240\001\001"
+ "eDescriptor\022\030\n\020bulkload_seq_num\030\004 \002(\003\022\023\n" +
+ "\013cluster_ids\030\005 \003(\t\"\272\002\n\025RegionEventDescri" +
+ "ptor\022=\n\nevent_type\030\001 \002(\0162).hbase.pb.Regi" +
+ "onEventDescriptor.EventType\022\022\n\ntable_nam",
+ "e\030\002 \002(\014\022\033\n\023encoded_region_name\030\003 \002(\014\022\033\n\023" +
+ "log_sequence_number\030\004 \001(\004\022)\n\006stores\030\005 \003(" +
+ "\0132\031.hbase.pb.StoreDescriptor\022$\n\006server\030\006" +
+ " \001(\0132\024.hbase.pb.ServerName\022\023\n\013region_nam" +
+ "e\030\007 \001(\014\".\n\tEventType\022\017\n\013REGION_OPEN\020\000\022\020\n" +
+ "\014REGION_CLOSE\020\001\"\014\n\nWALTrailer*F\n\tScopeTy" +
+ "pe\022\033\n\027REPLICATION_SCOPE_LOCAL\020\000\022\034\n\030REPLI" +
+ "CATION_SCOPE_GLOBAL\020\001B?\n*org.apache.hado" +
+ "op.hbase.protobuf.generatedB\tWALProtosH\001" +
+ "\210\001\000\240\001\001"
};
com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
@@ -12071,7 +12263,7 @@ public final class WALProtos {
internal_static_hbase_pb_BulkLoadDescriptor_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_hbase_pb_BulkLoadDescriptor_descriptor,
- new java.lang.String[] { "TableName", "EncodedRegionName", "Stores", "BulkloadSeqNum", });
+ new java.lang.String[] { "TableName", "EncodedRegionName", "Stores", "BulkloadSeqNum", "ClusterIds", });
internal_static_hbase_pb_RegionEventDescriptor_descriptor =
getDescriptor().getMessageTypes().get(7);
internal_static_hbase_pb_RegionEventDescriptor_fieldAccessorTable = new
diff --git a/hbase-protocol/src/main/protobuf/Client.proto b/hbase-protocol/src/main/protobuf/Client.proto
index 56f9f9f..b425bd4 100644
--- a/hbase-protocol/src/main/protobuf/Client.proto
+++ b/hbase-protocol/src/main/protobuf/Client.proto
@@ -343,6 +343,7 @@ message BulkLoadHFileRequest {
required RegionSpecifier region = 1;
repeated FamilyPath family_path = 2;
optional bool assign_seq_num = 3;
+ repeated string cluster_ids = 4;
message FamilyPath {
required bytes family = 1;
diff --git a/hbase-protocol/src/main/protobuf/SecureBulkLoad.proto b/hbase-protocol/src/main/protobuf/SecureBulkLoad.proto
index 814735b..cbbad37 100644
--- a/hbase-protocol/src/main/protobuf/SecureBulkLoad.proto
+++ b/hbase-protocol/src/main/protobuf/SecureBulkLoad.proto
@@ -31,6 +31,7 @@ message SecureBulkLoadHFilesRequest {
optional bool assign_seq_num = 2;
required DelegationToken fs_token = 3;
required string bulk_token = 4;
+ repeated string cluster_ids = 5;
}
message SecureBulkLoadHFilesResponse {
diff --git a/hbase-protocol/src/main/protobuf/WAL.proto b/hbase-protocol/src/main/protobuf/WAL.proto
index 2061b22..cc6a066 100644
--- a/hbase-protocol/src/main/protobuf/WAL.proto
+++ b/hbase-protocol/src/main/protobuf/WAL.proto
@@ -143,6 +143,7 @@ message BulkLoadDescriptor {
required bytes encoded_region_name = 2;
repeated StoreDescriptor stores = 3;
required int64 bulkload_seq_num = 4;
+ repeated string cluster_ids = 5;
}
/**
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/LoadIncrementalHFiles.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/LoadIncrementalHFiles.java
index 853b59d..7015b38 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/LoadIncrementalHFiles.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/LoadIncrementalHFiles.java
@@ -136,6 +136,8 @@ public class LoadIncrementalHFiles extends Configured implements Tool {
private UserProvider userProvider;
private int nrThreads;
+ private List<String> clusterIds = new ArrayList<>();
+
private LoadIncrementalHFiles() {}
public LoadIncrementalHFiles(Configuration conf) throws Exception {
@@ -143,6 +145,10 @@ public class LoadIncrementalHFiles extends Configured implements Tool {
initialize();
}
+ public void setClusterIds(List<String> clusterIds) {
+ this.clusterIds = clusterIds;
+ }
+
private void initialize() throws Exception {
if (hbAdmin == null) {
// make a copy, just to be sure we're not overriding someone else's config
@@ -854,7 +860,8 @@ public class LoadIncrementalHFiles extends Configured implements Tool {
+ Bytes.toStringBinary(getRow()) + " with hfile group " + famPaths);
byte[] regionName = getLocation().getRegionInfo().getRegionName();
if (!isSecureBulkLoadEndpointAvailable()) {
- success = ProtobufUtil.bulkLoadHFile(getStub(), famPaths, regionName, assignSeqIds);
+ success = ProtobufUtil.bulkLoadHFile(getStub(), famPaths, regionName, assignSeqIds,
+ clusterIds);
} else {
try (Table table = conn.getTable(getTableName())) {
secureClient = new SecureBulkLoadClient(table);
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
index 5a930d6..161a7a9 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
@@ -5700,7 +5700,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
@Override
public boolean bulkLoadHFiles(Collection<Pair<byte[], String>> familyPaths, boolean assignSeqId,
- BulkLoadListener bulkLoadListener) throws IOException {
+ BulkLoadListener bulkLoadListener, List<String> clusterIds) throws IOException {
long seqId = -1;
Map<byte[], List<Path>> storeFiles = new TreeMap<byte[], List<Path>>(Bytes.BYTES_COMPARATOR);
Map<String, Long> storeFilesSizes = new HashMap<String, Long>();
@@ -5837,7 +5837,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
WALProtos.BulkLoadDescriptor loadDescriptor =
ProtobufUtil.toBulkLoadDescriptor(this.getRegionInfo().getTable(),
ByteStringer.wrap(this.getRegionInfo().getEncodedNameAsBytes()), storeFiles,
- storeFilesSizes, seqId);
+ storeFilesSizes, seqId, clusterIds);
WALUtil.writeBulkLoadMarkerAndSync(wal, this.htableDescriptor, getRegionInfo(),
loadDescriptor, mvcc);
} catch (IOException ioe) {
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
index 5ce2796..2fdbd10 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
@@ -439,7 +439,7 @@ public class HRegionServer extends HasThread implements
/**
* Unique identifier for the cluster we are a part of.
*/
- private String clusterId;
+ String clusterId;
/**
* MX Bean for RegionServerInfo
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java
index 00fbdc9..d2b786d 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java
@@ -2114,6 +2114,12 @@ public class RSRpcServices implements HBaseRPCErrorHandler,
@Override
public BulkLoadHFileResponse bulkLoadHFile(final RpcController controller,
final BulkLoadHFileRequest request) throws ServiceException {
+ List<String> clusterIds = new ArrayList<String>(request.getClusterIdsList());
+ if(clusterIds.contains(this.regionServer.clusterId)){
+ return BulkLoadHFileResponse.newBuilder().setLoaded(true).build();
+ } else {
+ clusterIds.add(this.regionServer.clusterId);
+ }
try {
checkOpen();
requestCount.increment();
@@ -2128,11 +2134,14 @@ public class RSRpcServices implements HBaseRPCErrorHandler,
bypass = region.getCoprocessorHost().preBulkLoadHFile(familyPaths);
}
boolean loaded = false;
- if (!bypass) {
- loaded = region.bulkLoadHFiles(familyPaths, request.getAssignSeqNum(), null);
- }
- if (region.getCoprocessorHost() != null) {
- loaded = region.getCoprocessorHost().postBulkLoadHFile(familyPaths, loaded);
+ try {
+ if (!bypass) {
+ loaded = region.bulkLoadHFiles(familyPaths, request.getAssignSeqNum(), null, clusterIds);
+ }
+ } finally {
+ if (region.getCoprocessorHost() != null) {
+ loaded = region.getCoprocessorHost().postBulkLoadHFile(familyPaths, loaded);
+ }
}
BulkLoadHFileResponse.Builder builder = BulkLoadHFileResponse.newBuilder();
builder.setLoaded(loaded);
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Region.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Region.java
index 672d6ec..90e8c87 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Region.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Region.java
@@ -537,14 +537,15 @@ public interface Region extends ConfigurationObserver {
* rows with multiple column families atomically.
*
* @param familyPaths List of Pair<byte[] column family, String hfilePath>
+ * @param assignSeqId
* @param bulkLoadListener Internal hooks enabling massaging/preparation of a
* file about to be bulk loaded
- * @param assignSeqId
+ * @param clusterIds
* @return true if successful, false if failed recoverably
* @throws IOException if failed unrecoverably.
*/
boolean bulkLoadHFiles(Collection<Pair<byte[], String>> familyPaths, boolean assignSeqId,
- BulkLoadListener bulkLoadListener) throws IOException;
+ BulkLoadListener bulkLoadListener, List<String> clusterIds) throws IOException;
///////////////////////////////////////////////////////////////////////////
// Coprocessors
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HFileReplicator.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HFileReplicator.java
index 08606eb..84bc344 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HFileReplicator.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HFileReplicator.java
@@ -86,17 +86,19 @@ public class HFileReplicator {
private ThreadPoolExecutor exec;
private int maxCopyThreads;
private int copiesPerThread;
+ private List<String> sourceClusterIds;
public HFileReplicator(Configuration sourceClusterConf,
String sourceBaseNamespaceDirPath, String sourceHFileArchiveDirPath,
Map<String, List<Pair<byte[], List<String>>>> tableQueueMap, Configuration conf,
- Connection connection) throws IOException {
+ Connection connection, List<String> sourceClusterIds) throws IOException {
this.sourceClusterConf = sourceClusterConf;
this.sourceBaseNamespaceDirPath = sourceBaseNamespaceDirPath;
this.sourceHFileArchiveDirPath = sourceHFileArchiveDirPath;
this.bulkLoadHFileMap = tableQueueMap;
this.conf = conf;
this.connection = connection;
+ this.sourceClusterIds = sourceClusterIds;
userProvider = UserProvider.instantiate(conf);
fsDelegationToken = new FsDelegationToken(userProvider, "renewer");
@@ -130,6 +132,7 @@ public class HFileReplicator {
LoadIncrementalHFiles loadHFiles = null;
try {
loadHFiles = new LoadIncrementalHFiles(conf);
+ loadHFiles.setClusterIds(sourceClusterIds);
} catch (Exception e) {
LOG.error("Failed to initialize LoadIncrementalHFiles for replicating bulk loaded"
+ " data.", e);
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java
index 8f262c5..1f927d1 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java
@@ -155,9 +155,7 @@ public class ReplicationSink {
Map<TableName, Map<List<UUID>, List<Row>>> rowMap =
new TreeMap<TableName, Map<List<UUID>, List<Row>>>();
- // Map of table name Vs list of pair of family and list of hfile paths from its namespace
- Map<String, List<Pair<byte[], List<String>>>> bulkLoadHFileMap = null;
-
+ Map<List<String>, Map<String, List<Pair<byte[], List<String>>>>> bulkLoadsPerClusters = null;
for (WALEntry entry : entries) {
TableName table =
TableName.valueOf(entry.getKey().getTableName().toByteArray());
@@ -172,10 +170,19 @@ public class ReplicationSink {
Cell cell = cells.current();
// Handle bulk load hfiles replication
if (CellUtil.matchingQualifier(cell, WALEdit.BULK_LOAD)) {
+ BulkLoadDescriptor bld = WALEdit.getBulkLoadDescriptor(cell);
+ if(bulkLoadsPerClusters == null) {
+ bulkLoadsPerClusters = new HashMap<>();
+ }
+ // Map of table name Vs list of pair of family and list of
+ // hfile paths from its namespace
+ Map<String, List<Pair<byte[], List<String>>>> bulkLoadHFileMap =
+ bulkLoadsPerClusters.get(bld.getClusterIdsList());
if (bulkLoadHFileMap == null) {
- bulkLoadHFileMap = new HashMap<String, List<Pair<byte[], List<String>>>>();
+ bulkLoadHFileMap = new HashMap<>();
+ bulkLoadsPerClusters.put(bld.getClusterIdsList(), bulkLoadHFileMap);
}
- buildBulkLoadHFileMap(bulkLoadHFileMap, table, cell);
+ buildBulkLoadHFileMap(bulkLoadHFileMap, table, bld);
} else {
// Handle wal replication
if (isNewRowOrType(previousCell, cell)) {
@@ -211,14 +218,26 @@ public class ReplicationSink {
LOG.debug("Finished replicating mutations.");
}
- if (bulkLoadHFileMap != null && !bulkLoadHFileMap.isEmpty()) {
- LOG.debug("Started replicating bulk loaded data.");
- HFileReplicator hFileReplicator =
- new HFileReplicator(this.provider.getConf(this.conf, replicationClusterId),
+ if(bulkLoadsPerClusters != null) {
+ for (Entry<List<String>, Map<String, List<Pair<byte[],
+ List<String>>>>> entry : bulkLoadsPerClusters.entrySet()) {
+ Map<String, List<Pair<byte[], List<String>>>> bulkLoadHFileMap = entry.getValue();
+ if (bulkLoadHFileMap != null && !bulkLoadHFileMap.isEmpty()) {
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Started replicating bulk loaded data from cluster ids: " +
+ entry.getKey().toString());
+ }
+ HFileReplicator hFileReplicator =
+ new HFileReplicator(this.provider.getConf(this.conf, replicationClusterId),
sourceBaseNamespaceDirPath, sourceHFileArchiveDirPath, bulkLoadHFileMap, conf,
- getConnection());
- hFileReplicator.replicate();
- LOG.debug("Finished replicating bulk loaded data.");
+ getConnection(), entry.getKey());
+ hFileReplicator.replicate();
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Finished replicating bulk loaded data from cluster id: " +
+ entry.getKey().toString());
+ }
+ }
+ }
}
int size = entries.size();
@@ -233,8 +252,7 @@ public class ReplicationSink {
private void buildBulkLoadHFileMap(
final Map<String, List<Pair<byte[], List<String>>>> bulkLoadHFileMap, TableName table,
- Cell cell) throws IOException {
- BulkLoadDescriptor bld = WALEdit.getBulkLoadDescriptor(cell);
+ BulkLoadDescriptor bld) throws IOException {
List<StoreDescriptor> storesList = bld.getStoresList();
int storesSize = storesList.size();
for (int j = 0; j < storesSize; j++) {
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/SecureBulkLoadEndpoint.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/SecureBulkLoadEndpoint.java
index 6582a35..170a6f2 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/SecureBulkLoadEndpoint.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/SecureBulkLoadEndpoint.java
@@ -294,7 +294,7 @@ public class SecureBulkLoadEndpoint extends SecureBulkLoadService
@Override
public void secureBulkLoadHFiles(RpcController controller,
- SecureBulkLoadHFilesRequest request,
+ final SecureBulkLoadHFilesRequest request,
RpcCallback<SecureBulkLoadHFilesResponse> done) {
final List<Pair<byte[], String>> familyPaths = new ArrayList<Pair<byte[], String>>();
for(ClientProtos.BulkLoadHFileRequest.FamilyPath el : request.getFamilyPathList()) {
@@ -376,7 +376,7 @@ public class SecureBulkLoadEndpoint extends SecureBulkLoadService
//We call bulkLoadHFiles as requesting user
//To enable access prior to staging
return env.getRegion().bulkLoadHFiles(familyPaths, true,
- new SecureBulkLoadListener(fs, bulkToken, conf));
+ new SecureBulkLoadListener(fs, bulkToken, conf), request.getClusterIdsList());
} catch (Exception e) {
LOG.error("Failed to complete bulk load", e);
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestBulkLoad.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestBulkLoad.java
index 735fec8..77f376c 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestBulkLoad.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestBulkLoad.java
@@ -122,14 +122,14 @@ public class TestBulkLoad {
};
});
testRegionWithFamiliesAndSpecifiedTableName(tableName, family1)
- .bulkLoadHFiles(familyPaths, false, null);
+ .bulkLoadHFiles(familyPaths, false, null, null);
verify(log).sync(anyLong());
}
@Test
public void bulkHLogShouldThrowNoErrorAndWriteMarkerWithBlankInput() throws IOException {
testRegionWithFamilies(family1).bulkLoadHFiles(new ArrayList<Pair<byte[], String>>(),
- false, null);
+ false, null, null);
}
@Test
@@ -147,7 +147,7 @@ public class TestBulkLoad {
return 01L;
};
});
- testRegionWithFamilies(family1).bulkLoadHFiles(withFamilyPathsFor(family1), false, null);
+ testRegionWithFamilies(family1).bulkLoadHFiles(withFamilyPathsFor(family1), false, null, null);
verify(log).sync(anyLong());
}
@@ -167,7 +167,7 @@ public class TestBulkLoad {
};
});
testRegionWithFamilies(family1, family2).bulkLoadHFiles(withFamilyPathsFor(family1, family2),
- false, null);
+ false, null, null);
verify(log).sync(anyLong());
}
@@ -188,33 +188,33 @@ public class TestBulkLoad {
});
TableName tableName = TableName.valueOf("test", "test");
testRegionWithFamiliesAndSpecifiedTableName(tableName, family1, family2)
- .bulkLoadHFiles(withFamilyPathsFor(family1, family2), false, null);
+ .bulkLoadHFiles(withFamilyPathsFor(family1, family2), false, null, null);
verify(log).sync(anyLong());
}
@Test(expected = DoNotRetryIOException.class)
public void shouldCrashIfBulkLoadFamiliesNotInTable() throws IOException {
testRegionWithFamilies(family1).bulkLoadHFiles(withFamilyPathsFor(family1, family2), false,
- null);
+ null, null);
}
@Test(expected = DoNotRetryIOException.class)
public void bulkHLogShouldThrowErrorWhenFamilySpecifiedAndHFileExistsButNotInTableDescriptor()
throws IOException {
- testRegionWithFamilies().bulkLoadHFiles(withFamilyPathsFor(family1), false, null);
+ testRegionWithFamilies().bulkLoadHFiles(withFamilyPathsFor(family1), false, null, null);
}
@Test(expected = DoNotRetryIOException.class)
public void shouldThrowErrorIfBadFamilySpecifiedAsFamilyPath() throws IOException {
testRegionWithFamilies()
.bulkLoadHFiles(asList(withInvalidColumnFamilyButProperHFileLocation(family1)),
- false, null);
+ false, null, null);
}
@Test(expected = FileNotFoundException.class)
public void shouldThrowErrorIfHFileDoesNotExist() throws IOException {
List<Pair<byte[], String>> list = asList(withMissingHFileForFamily(family1));
- testRegionWithFamilies(family1).bulkLoadHFiles(list, false, null);
+ testRegionWithFamilies(family1).bulkLoadHFiles(list, false, null, null);
}
private Pair<byte[], String> withMissingHFileForFamily(byte[] family) {
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestBulkLoadReplication.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestBulkLoadReplication.java
new file mode 100644
index 0000000..6754e8d
--- /dev/null
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestBulkLoadReplication.java
@@ -0,0 +1,298 @@
+/**
+ * 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.hadoop.hbase.regionserver;
+
+import static org.apache.hadoop.hbase.HConstants.REPLICATION_CLUSTER_ID;
+import static org.apache.hadoop.hbase.HConstants.REPLICATION_CONF_DIR;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.ConnectionFactory;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.client.replication.ReplicationAdmin;
+import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
+import org.apache.hadoop.hbase.coprocessor.ObserverContext;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.io.hfile.HFile;
+import org.apache.hadoop.hbase.io.hfile.HFileContext;
+import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
+import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
+import org.apache.hadoop.hbase.replication.TestReplicationBase;
+import org.apache.hadoop.hbase.testclassification.MediumTests;
+import org.apache.hadoop.hbase.testclassification.ReplicationTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.Pair;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TemporaryFolder;
+import org.junit.rules.TestName;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Integration test for bulk load replication. Defines three clusters, with the following
+ * replication topology: "1 <-> 2 <-> 3" (active-active between 1 and 2, and active-active between
+ * 2 and 3).
+ *
+ * For each of defined test clusters, it performs a bulk load, asserting values on bulk loaded file
+ * gets replicated to other two peers. Since we are doing 3 bulk loads, with the given replication
+ * topology all these bulk loads should get replicated only once on each peer. To assert this,
+ * this test defines a preBulkLoad coprocessor and adds it to all test table regions, on each of the
+ * clusters. This CP counts the amount of times bulk load actually gets invoked, certifying
+ * we are not entering the infinite loop condition addressed by HBASE-22380.
+ */
+@Category({ ReplicationTests.class, MediumTests.class})
+public class TestBulkLoadReplication extends TestReplicationBase {
+
+ protected static final Logger LOG =
+ LoggerFactory.getLogger(TestBulkLoadReplication.class);
+
+ private static final String PEER1_CLUSTER_ID = "peer1";
+ private static final String PEER4_CLUSTER_ID = "peer4";
+ private static final String PEER3_CLUSTER_ID = "peer3";
+
+ private static final String PEER_ID1 = "1";
+ private static final String PEER_ID3 = "3";
+ private static final String PEER_ID4 = "4";
+
+ private static final AtomicInteger BULK_LOADS_COUNT = new AtomicInteger(0);
+ private static CountDownLatch BULK_LOAD_LATCH;
+
+ private static HBaseTestingUtility utility3;
+ private static HBaseTestingUtility utility4;
+ private static Configuration conf3;
+ private static Configuration conf4;
+// private static Table htable3;
+// private static Table htable4;
+
+ @Rule
+ public TestName name = new TestName();
+
+ @ClassRule
+ public static TemporaryFolder testFolder = new TemporaryFolder();
+
+ @BeforeClass
+ public static void setUpBeforeClass() throws Exception {
+ setupBulkLoadConfigsForCluster(conf1, PEER1_CLUSTER_ID);
+ conf3 = HBaseConfiguration.create(conf1);
+ setupBulkLoadConfigsForCluster(conf3, PEER3_CLUSTER_ID);
+ conf3.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/3");
+ utility3 = new HBaseTestingUtility(conf3);
+ conf4 = HBaseConfiguration.create(conf1);
+ setupBulkLoadConfigsForCluster(conf4, PEER4_CLUSTER_ID);
+ conf4.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/4");
+ utility4 = new HBaseTestingUtility(conf4);
+ TestReplicationBase.setUpBeforeClass();
+ startCluster(utility3, conf3);
+ startCluster(utility4, conf4);
+ }
+
+ private static void startCluster(HBaseTestingUtility util, Configuration configuration)
+ throws Exception {
+ LOG.info("Setup Zk to same one from utility1 and utility4");
+ util.setZkCluster(utility1.getZkCluster());
+ util.startMiniCluster(2);
+
+ HTableDescriptor tableDesc = new HTableDescriptor(tableName);
+ HColumnDescriptor columnDesc = new HColumnDescriptor(famName);
+ columnDesc.setScope(1);
+ tableDesc.addFamily(columnDesc);
+
+ Connection connection = ConnectionFactory.createConnection(configuration);
+ try (Admin admin = connection.getAdmin()) {
+ admin.createTable(tableDesc, HBaseTestingUtility.KEYS_FOR_HBA_CREATE_TABLE);
+ }
+ util.waitUntilAllRegionsAssigned(tableName);
+ }
+
+ @Before
+ public void setUpBase() throws Exception {
+ ReplicationPeerConfig peer1Config = getPeerConfigForCluster(utility1);
+ ReplicationPeerConfig peer4Config = getPeerConfigForCluster(utility4);
+ ReplicationPeerConfig peer3Config = getPeerConfigForCluster(utility3);
+ //adds cluster4 as a remote peer on cluster1
+ getReplicationAdmin(utility1.getConfiguration()).addPeer(PEER_ID4, peer4Config, null);
+ //adds cluster1 as a remote peer on cluster4
+ ReplicationAdmin admin4 = getReplicationAdmin(utility4.getConfiguration());
+ admin4.addPeer(PEER_ID1, peer1Config, null);
+ //adds cluster3 as a remote peer on cluster4
+ admin4.addPeer(PEER_ID3, peer3Config, null);
+ //adds cluster4 as a remote peer on cluster3
+ getReplicationAdmin(utility3.getConfiguration()).addPeer(PEER_ID4, peer4Config, null);
+ setupCoprocessor(utility1);
+ setupCoprocessor(utility4);
+ setupCoprocessor(utility3);
+ }
+
+ private ReplicationAdmin getReplicationAdmin(Configuration configuration) throws IOException {
+ return new ReplicationAdmin(configuration);
+ }
+
+ private ReplicationPeerConfig getPeerConfigForCluster(HBaseTestingUtility util) {
+ ReplicationPeerConfig config = new ReplicationPeerConfig();
+ config.setClusterKey(util.getClusterKey());
+ return config;
+ }
+
+ private void setupCoprocessor(HBaseTestingUtility cluster) throws IOException {
+ for(HRegion region : cluster.getHBaseCluster().getRegions(tableName)){
+ region.getCoprocessorHost().load(TestBulkLoadReplication.BulkReplicationTestObserver.class,
+ 0, cluster.getConfiguration());
+ }
+ }
+
+ @After
+ public void tearDownBase() throws Exception {
+ getReplicationAdmin(utility4.getConfiguration()).removePeer(PEER_ID1);
+ getReplicationAdmin(utility4.getConfiguration()).removePeer(PEER_ID3);
+ getReplicationAdmin(utility3.getConfiguration()).removePeer(PEER_ID4);
+ }
+
+ private static void setupBulkLoadConfigsForCluster(Configuration config,
+ String clusterReplicationId) throws Exception {
+ config.setBoolean(HConstants.REPLICATION_BULKLOAD_ENABLE_KEY, true);
+ config.set(REPLICATION_CLUSTER_ID, clusterReplicationId);
+ File sourceConfigFolder = testFolder.newFolder(clusterReplicationId);
+ File sourceConfigFile = new File(sourceConfigFolder.getAbsolutePath()
+ + "/hbase-site.xml");
+ config.writeXml(new FileOutputStream(sourceConfigFile));
+ config.set(REPLICATION_CONF_DIR, testFolder.getRoot().getAbsolutePath());
+ }
+
+ @Test
+ public void testBulkLoadReplicationActiveActive() throws Exception {
+ Table peer1TestTable = utility1.getConnection().getTable(TestReplicationBase.tableName);
+ Table peer4TestTable = utility4.getConnection().getTable(TestReplicationBase.tableName);
+ Table peer3TestTable = utility3.getConnection().getTable(TestReplicationBase.tableName);
+ byte[] row = Bytes.toBytes("001");
+ byte[] value = Bytes.toBytes("v1");
+ assertBulkLoadConditions(row, value, utility1, peer1TestTable, peer4TestTable, peer3TestTable);
+ row = Bytes.toBytes("002");
+ value = Bytes.toBytes("v2");
+ assertBulkLoadConditions(row, value, utility4, peer4TestTable, peer1TestTable, peer3TestTable);
+ row = Bytes.toBytes("003");
+ value = Bytes.toBytes("v3");
+ assertBulkLoadConditions(row, value, utility3, peer3TestTable, peer4TestTable, peer1TestTable);
+ //Additional wait to make sure no extra bulk load happens
+ Thread.sleep(400);
+ //We have 3 bulk load events (1 initiated on each cluster).
+ //Each event gets 3 counts (the originator cluster, plus the two peers),
+ //so BULK_LOADS_COUNT expected value is 3 * 3 = 9.
+ assertEquals(9, BULK_LOADS_COUNT.get());
+ }
+
+ private void assertBulkLoadConditions(byte[] row, byte[] value,
+ HBaseTestingUtility utility, Table...tables) throws Exception {
+ BULK_LOAD_LATCH = new CountDownLatch(3);
+ bulkLoadOnCluster(row, value, utility);
+ assertTrue(BULK_LOAD_LATCH.await(1, TimeUnit.MINUTES));
+ assertTableHasValue(tables[0], row, value);
+ assertTableHasValue(tables[1], row, value);
+ assertTableHasValue(tables[2], row, value);
+ }
+
+ private void bulkLoadOnCluster(byte[] row, byte[] value,
+ HBaseTestingUtility cluster) throws Exception {
+ String bulkLoadFile = createHFileForFamilies(row, value, cluster.getConfiguration());
+ copyToHdfs(bulkLoadFile, cluster.getDFSCluster());
+ LoadIncrementalHFiles bulkLoadHFilesTool =
+ new LoadIncrementalHFiles(cluster.getConfiguration());
+ bulkLoadHFilesTool.run(new String[]{"/bulk_dir/region1/", tableName.getNameAsString()});
+ }
+
+ private void copyToHdfs(String bulkLoadFilePath, MiniDFSCluster cluster) throws Exception {
+ Path bulkLoadDir = new Path("/bulk_dir/region1/f");
+ cluster.getFileSystem().mkdirs(bulkLoadDir);
+ cluster.getFileSystem().copyFromLocalFile(new Path(bulkLoadFilePath), bulkLoadDir);
+ }
+
+ private void assertTableHasValue(Table table, byte[] row, byte[] value) throws Exception {
+ Get get = new Get(row);
+ Result result = table.get(get);
+ assertTrue(result.advance());
+ assertEquals(Bytes.toString(value), Bytes.toString(result.value()));
+ }
+
+ private String createHFileForFamilies(byte[] row, byte[] value,
+ Configuration clusterConfig) throws IOException {
+ final KeyValue kv = new KeyValue(row, famName, Bytes.toBytes("1"), System.currentTimeMillis(),
+ KeyValue.Type.Put, value);
+ final HFile.WriterFactory hFileFactory = HFile.getWriterFactoryNoCache(clusterConfig);
+ // TODO We need a way to do this without creating files
+ final File hFileLocation = testFolder.newFile();
+ final FSDataOutputStream out =
+ new FSDataOutputStream(new FileOutputStream(hFileLocation), null);
+ try {
+ hFileFactory.withOutputStream(out);
+ hFileFactory.withFileContext(new HFileContext());
+ HFile.Writer writer = hFileFactory.create();
+ try {
+ writer.append(kv);
+ } finally {
+ writer.close();
+ }
+ } finally {
+ out.close();
+ }
+ return hFileLocation.getAbsoluteFile().getAbsolutePath();
+ }
+
+ public static class BulkReplicationTestObserver extends BaseRegionObserver {
+
+ @Override
+ public void preBulkLoadHFile(ObserverContext<RegionCoprocessorEnvironment> ctx,
+ List<Pair<byte[], String>> familyPaths) throws IOException {
+ BULK_LOADS_COUNT.incrementAndGet();
+ }
+
+ @Override
+ public boolean postBulkLoadHFile(ObserverContext<RegionCoprocessorEnvironment> ctx,
+ List<Pair<byte[], String>> familyPaths, boolean hasLoaded) throws IOException {
+ if(hasLoaded) {
+ BULK_LOAD_LATCH.countDown();
+ }
+ return true;
+ }
+
+ }
+}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionReplayEvents.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionReplayEvents.java
index 3a8b557..481e5c9 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionReplayEvents.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionReplayEvents.java
@@ -1480,7 +1480,7 @@ public class TestHRegionReplayEvents {
randomValues)));
expectedLoadFileCount++;
}
- primaryRegion.bulkLoadHFiles(familyPaths, false, null);
+ primaryRegion.bulkLoadHFiles(familyPaths, false, null, null);
// now replay the edits and the bulk load marker
reader = createWALReaderForPrimary();
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplay.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplay.java
index 8758c98..c922145 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplay.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplay.java
@@ -383,7 +383,7 @@ public class TestWALReplay {
Bytes.toBytes("z"), 10);
List <Pair<byte[],String>> hfs= new ArrayList<Pair<byte[],String>>(1);
hfs.add(Pair.newPair(family, f.toString()));
- region.bulkLoadHFiles(hfs, true, null);
+ region.bulkLoadHFiles(hfs, true, null, null);
// Add an edit so something in the WAL
byte [] row = tableName.getName();
@@ -457,7 +457,7 @@ public class TestWALReplay {
Bytes.toBytes(i + "50"), 10);
hfs.add(Pair.newPair(family, f.toString()));
}
- region.bulkLoadHFiles(hfs, true, null);
+ region.bulkLoadHFiles(hfs, true, null, null);
final int rowsInsertedCount = 31;
assertEquals(rowsInsertedCount, getScannedCount(region.getScanner(new Scan())));