View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
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     * A helper to replicate a list of WAL entries using admin protocol.
56     * @param admin Admin service
57     * @param entries Array of WAL entries to be replicated
58     * @param replicationClusterId Id which will uniquely identify source cluster FS client
59     *          configurations in the replication configuration directory
60     * @param sourceBaseNamespaceDir Path to source cluster base namespace directory
61     * @param sourceHFileArchiveDir Path to the source cluster hfile archive directory
62     * @throws java.io.IOException
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     * Create a new ReplicateWALEntryRequest from a list of WAL entries
80     *
81     * @param entries the WAL entries to be replicated
82     * @return a pair of ReplicateWALEntryRequest and a CellScanner over all the WALEdit values
83     * found.
84     */
85    public static Pair<AdminProtos.ReplicateWALEntryRequest, CellScanner>
86        buildReplicateWALEntryRequest(final Entry[] entries) {
87      // Accumulate all the Cells seen in here.
88      return buildReplicateWALEntryRequest(entries, null, null, null, null);
89    }
90  
91    /**
92     * Create a new ReplicateWALEntryRequest from a list of HLog entries
93     * @param entries the HLog entries to be replicated
94     * @param encodedRegionName alternative region name to use if not null
95     * @param replicationClusterId Id which will uniquely identify source cluster FS client
96     *          configurations in the replication configuration directory
97     * @param sourceBaseNamespaceDir Path to source cluster base namespace directory
98     * @param sourceHFileArchiveDir Path to the source cluster hfile archive directory
99     * @return a pair of ReplicateWALEntryRequest and a CellScanner over all the WALEdit values found.
100    */
101   public static Pair<AdminProtos.ReplicateWALEntryRequest, CellScanner>
102       buildReplicateWALEntryRequest(final Entry[] entries, byte[] encodedRegionName,
103           String replicationClusterId, Path sourceBaseNamespaceDir, Path sourceHFileArchiveDir) {
104     // Accumulate all the KVs seen in here.
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       // TODO: this duplicates a lot in WALKey#getBuilder
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       // Add up the size.  It is used later serializing out the cells.
151       for (Cell cell: cells) {
152         size += CellUtil.estimatedSerializedSizeOf(cell);
153       }
154       // Collect up the cells
155       allCells.add(cells);
156       // Write out how many cells associated with this entry.
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    * @param cells
177    * @return <code>cells</code> packaged as a CellScanner
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 }