1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.coprocessor;
19
20 import java.io.IOException;
21 import java.util.ArrayList;
22 import java.util.List;
23 import java.util.SortedSet;
24 import java.util.TreeSet;
25
26 import org.apache.hadoop.hbase.classification.InterfaceAudience;
27 import org.apache.hadoop.hbase.classification.InterfaceStability;
28 import org.apache.hadoop.hbase.Coprocessor;
29 import org.apache.hadoop.hbase.CoprocessorEnvironment;
30 import org.apache.hadoop.hbase.HBaseInterfaceAudience;
31 import org.apache.hadoop.hbase.HConstants;
32 import org.apache.hadoop.hbase.HRegionInfo;
33 import org.apache.hadoop.hbase.client.Mutation;
34 import org.apache.hadoop.hbase.regionserver.HRegion;
35 import org.apache.hadoop.hbase.regionserver.WrongRegionException;
36 import org.apache.hadoop.hbase.util.Bytes;
37 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
38 import org.apache.hadoop.hbase.protobuf.ResponseConverter;
39 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto;
40 import org.apache.hadoop.hbase.protobuf.generated.MultiRowMutationProtos.MutateRowsRequest;
41 import org.apache.hadoop.hbase.protobuf.generated.MultiRowMutationProtos.MutateRowsResponse;
42 import org.apache.hadoop.hbase.protobuf.generated.MultiRowMutationProtos.MultiRowMutationService;
43
44 import com.google.protobuf.RpcCallback;
45 import com.google.protobuf.RpcController;
46 import com.google.protobuf.Service;
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78 @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.COPROC)
79 @InterfaceStability.Evolving
80 public class MultiRowMutationEndpoint extends MultiRowMutationService implements
81 CoprocessorService, Coprocessor {
82 private RegionCoprocessorEnvironment env;
83 @Override
84 public void mutateRows(RpcController controller, MutateRowsRequest request,
85 RpcCallback<MutateRowsResponse> done) {
86 MutateRowsResponse response = MutateRowsResponse.getDefaultInstance();
87 try {
88
89 SortedSet<byte[]> rowsToLock = new TreeSet<byte[]>(Bytes.BYTES_COMPARATOR);
90 List<MutationProto> mutateRequestList = request.getMutationRequestList();
91 List<Mutation> mutations = new ArrayList<Mutation>(mutateRequestList.size());
92 for (MutationProto m : mutateRequestList) {
93 mutations.add(ProtobufUtil.toMutation(m));
94 }
95
96 HRegionInfo regionInfo = env.getRegion().getRegionInfo();
97 for (Mutation m : mutations) {
98
99 if (!HRegion.rowIsInRange(regionInfo, m.getRow())) {
100 String msg = "Requested row out of range '"
101 + Bytes.toStringBinary(m.getRow()) + "'";
102 if (rowsToLock.isEmpty()) {
103
104
105 throw new WrongRegionException(msg);
106 } else {
107
108 throw new org.apache.hadoop.hbase.DoNotRetryIOException(msg);
109 }
110 }
111 rowsToLock.add(m.getRow());
112 }
113
114 long nonceGroup = request.hasNonceGroup() ? request.getNonceGroup() : HConstants.NO_NONCE;
115 long nonce = request.hasNonce() ? request.getNonce() : HConstants.NO_NONCE;
116 env.getRegion().mutateRowsWithLocks(mutations, rowsToLock, nonceGroup, nonce);
117 } catch (IOException e) {
118 ResponseConverter.setControllerException(controller, e);
119 }
120 done.run(response);
121 }
122
123
124 @Override
125 public Service getService() {
126 return this;
127 }
128
129
130
131
132
133
134
135
136
137
138
139 @Override
140 public void start(CoprocessorEnvironment env) throws IOException {
141 if (env instanceof RegionCoprocessorEnvironment) {
142 this.env = (RegionCoprocessorEnvironment)env;
143 } else {
144 throw new CoprocessorException("Must be loaded on a table region!");
145 }
146 }
147
148 @Override
149 public void stop(CoprocessorEnvironment env) throws IOException {
150
151 }
152 }