1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20 package org.apache.hadoop.hbase.protobuf;
21
22
23 import com.google.protobuf.ServiceException;
24
25 import java.io.IOException;
26 import java.util.ArrayList;
27 import java.util.Iterator;
28 import java.util.List;
29 import java.util.Map;
30 import java.util.NavigableMap;
31 import java.util.UUID;
32
33 import org.apache.hadoop.fs.Path;
34 import org.apache.hadoop.hbase.Cell;
35 import org.apache.hadoop.hbase.CellScanner;
36 import org.apache.hadoop.hbase.CellUtil;
37 import org.apache.hadoop.hbase.HConstants;
38 import org.apache.hadoop.hbase.classification.InterfaceAudience;
39 import org.apache.hadoop.hbase.io.SizedCellScanner;
40 import org.apache.hadoop.hbase.ipc.HBaseRpcController;
41 import org.apache.hadoop.hbase.ipc.HBaseRpcControllerImpl;
42 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos;
43 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.AdminService;
44 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos;
45 import org.apache.hadoop.hbase.protobuf.generated.WALProtos;
46 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
47 import org.apache.hadoop.hbase.util.ByteStringer;
48 import org.apache.hadoop.hbase.util.Pair;
49 import org.apache.hadoop.hbase.wal.WAL.Entry;
50 import org.apache.hadoop.hbase.wal.WALKey;
51
52 @InterfaceAudience.Private
53 public class ReplicationProtbufUtil {
54
55
56
57
58
59
60
61
62
63
64 public static void replicateWALEntry(final AdminService.BlockingInterface admin,
65 final Entry[] entries, String replicationClusterId, Path sourceBaseNamespaceDir,
66 Path sourceHFileArchiveDir) throws IOException {
67 Pair<AdminProtos.ReplicateWALEntryRequest, CellScanner> p =
68 buildReplicateWALEntryRequest(entries, null, replicationClusterId, sourceBaseNamespaceDir,
69 sourceHFileArchiveDir);
70 HBaseRpcController controller = new HBaseRpcControllerImpl(p.getSecond());
71 try {
72 admin.replicateWALEntry(controller, p.getFirst());
73 } catch (ServiceException se) {
74 throw ProtobufUtil.getServiceException(se);
75 }
76 }
77
78
79
80
81
82
83
84
85 public static Pair<AdminProtos.ReplicateWALEntryRequest, CellScanner>
86 buildReplicateWALEntryRequest(final Entry[] entries) {
87
88 return buildReplicateWALEntryRequest(entries, null, null, null, null);
89 }
90
91
92
93
94
95
96
97
98
99
100
101 public static Pair<AdminProtos.ReplicateWALEntryRequest, CellScanner>
102 buildReplicateWALEntryRequest(final Entry[] entries, byte[] encodedRegionName,
103 String replicationClusterId, Path sourceBaseNamespaceDir, Path sourceHFileArchiveDir) {
104
105 List<List<? extends Cell>> allCells = new ArrayList<List<? extends Cell>>(entries.length);
106 int size = 0;
107 WALProtos.FamilyScope.Builder scopeBuilder = WALProtos.FamilyScope.newBuilder();
108 AdminProtos.WALEntry.Builder entryBuilder = AdminProtos.WALEntry.newBuilder();
109 AdminProtos.ReplicateWALEntryRequest.Builder builder =
110 AdminProtos.ReplicateWALEntryRequest.newBuilder();
111 HBaseProtos.UUID.Builder uuidBuilder = HBaseProtos.UUID.newBuilder();
112 for (Entry entry: entries) {
113 entryBuilder.clear();
114
115 WALProtos.WALKey.Builder keyBuilder = entryBuilder.getKeyBuilder();
116 WALKey key = entry.getKey();
117 keyBuilder.setEncodedRegionName(
118 ByteStringer.wrap(encodedRegionName == null
119 ? key.getEncodedRegionName()
120 : encodedRegionName));
121 keyBuilder.setTableName(ByteStringer.wrap(key.getTablename().getName()));
122 keyBuilder.setLogSequenceNumber(key.getLogSeqNum());
123 keyBuilder.setWriteTime(key.getWriteTime());
124 if (key.getNonce() != HConstants.NO_NONCE) {
125 keyBuilder.setNonce(key.getNonce());
126 }
127 if (key.getNonceGroup() != HConstants.NO_NONCE) {
128 keyBuilder.setNonceGroup(key.getNonceGroup());
129 }
130 for(UUID clusterId : key.getClusterIds()) {
131 uuidBuilder.setLeastSigBits(clusterId.getLeastSignificantBits());
132 uuidBuilder.setMostSigBits(clusterId.getMostSignificantBits());
133 keyBuilder.addClusterIds(uuidBuilder.build());
134 }
135 if(key.getOrigLogSeqNum() > 0) {
136 keyBuilder.setOrigSequenceNumber(key.getOrigLogSeqNum());
137 }
138 WALEdit edit = entry.getEdit();
139 NavigableMap<byte[], Integer> scopes = key.getScopes();
140 if (scopes != null && !scopes.isEmpty()) {
141 for (Map.Entry<byte[], Integer> scope: scopes.entrySet()) {
142 scopeBuilder.setFamily(ByteStringer.wrap(scope.getKey()));
143 WALProtos.ScopeType scopeType =
144 WALProtos.ScopeType.valueOf(scope.getValue().intValue());
145 scopeBuilder.setScopeType(scopeType);
146 keyBuilder.addScopes(scopeBuilder.build());
147 }
148 }
149 List<Cell> cells = edit.getCells();
150
151 for (Cell cell: cells) {
152 size += CellUtil.estimatedSerializedSizeOf(cell);
153 }
154
155 allCells.add(cells);
156
157 entryBuilder.setAssociatedCellCount(cells.size());
158 builder.addEntry(entryBuilder.build());
159 }
160
161 if (replicationClusterId != null) {
162 builder.setReplicationClusterId(replicationClusterId);
163 }
164 if (sourceBaseNamespaceDir != null) {
165 builder.setSourceBaseNamespaceDirPath(sourceBaseNamespaceDir.toString());
166 }
167 if (sourceHFileArchiveDir != null) {
168 builder.setSourceHFileArchiveDirPath(sourceHFileArchiveDir.toString());
169 }
170
171 return new Pair<AdminProtos.ReplicateWALEntryRequest, CellScanner>(builder.build(),
172 getCellScanner(allCells, size));
173 }
174
175
176
177
178
179 static CellScanner getCellScanner(final List<List<? extends Cell>> cells, final int size) {
180 return new SizedCellScanner() {
181 private final Iterator<List<? extends Cell>> entries = cells.iterator();
182 private Iterator<? extends Cell> currentIterator = null;
183 private Cell currentCell;
184
185 @Override
186 public Cell current() {
187 return this.currentCell;
188 }
189
190 @Override
191 public boolean advance() {
192 if (this.currentIterator == null) {
193 if (!this.entries.hasNext()) return false;
194 this.currentIterator = this.entries.next().iterator();
195 }
196 if (this.currentIterator.hasNext()) {
197 this.currentCell = this.currentIterator.next();
198 return true;
199 }
200 this.currentCell = null;
201 this.currentIterator = null;
202 return advance();
203 }
204
205 @Override
206 public long heapSize() {
207 return size;
208 }
209 };
210 }
211 }