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.regionserver;
21
22 import java.io.IOException;
23 import java.util.ArrayList;
24 import java.util.List;
25 import java.util.Map;
26 import java.util.NavigableSet;
27 import java.util.UUID;
28 import java.util.concurrent.ConcurrentHashMap;
29 import java.util.concurrent.ConcurrentMap;
30 import java.util.regex.Matcher;
31
32 import org.apache.commons.collections.map.AbstractReferenceMap;
33 import org.apache.commons.collections.map.ReferenceMap;
34 import org.apache.commons.lang.ClassUtils;
35 import org.apache.commons.logging.Log;
36 import org.apache.commons.logging.LogFactory;
37 import org.apache.hadoop.hbase.classification.InterfaceAudience;
38 import org.apache.hadoop.hbase.classification.InterfaceStability;
39 import org.apache.hadoop.conf.Configuration;
40 import org.apache.hadoop.fs.FileSystem;
41 import org.apache.hadoop.fs.Path;
42 import org.apache.hadoop.hbase.Cell;
43 import org.apache.hadoop.hbase.Coprocessor;
44 import org.apache.hadoop.hbase.CoprocessorEnvironment;
45 import org.apache.hadoop.hbase.HBaseConfiguration;
46 import org.apache.hadoop.hbase.HBaseInterfaceAudience;
47 import org.apache.hadoop.hbase.HConstants;
48 import org.apache.hadoop.hbase.HRegionInfo;
49 import org.apache.hadoop.hbase.HTableDescriptor;
50 import org.apache.hadoop.hbase.client.Append;
51 import org.apache.hadoop.hbase.client.Delete;
52 import org.apache.hadoop.hbase.client.Durability;
53 import org.apache.hadoop.hbase.client.Get;
54 import org.apache.hadoop.hbase.client.Increment;
55 import org.apache.hadoop.hbase.client.Mutation;
56 import org.apache.hadoop.hbase.client.Put;
57 import org.apache.hadoop.hbase.client.Result;
58 import org.apache.hadoop.hbase.client.Scan;
59 import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
60 import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
61 import org.apache.hadoop.hbase.coprocessor.CoprocessorService;
62 import org.apache.hadoop.hbase.coprocessor.EndpointObserver;
63 import org.apache.hadoop.hbase.coprocessor.MetricsCoprocessor;
64 import org.apache.hadoop.hbase.coprocessor.ObserverContext;
65 import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
66 import org.apache.hadoop.hbase.coprocessor.RegionObserver;
67 import org.apache.hadoop.hbase.coprocessor.RegionObserver.MutationType;
68 import org.apache.hadoop.hbase.filter.ByteArrayComparable;
69 import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
70 import org.apache.hadoop.hbase.io.FSDataInputStreamWrapper;
71 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
72 import org.apache.hadoop.hbase.io.Reference;
73 import org.apache.hadoop.hbase.io.hfile.CacheConfig;
74 import org.apache.hadoop.hbase.metrics.MetricRegistry;
75 import org.apache.hadoop.hbase.regionserver.DeleteTracker;
76 import org.apache.hadoop.hbase.regionserver.Region.Operation;
77 import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequest;
78 import org.apache.hadoop.hbase.regionserver.wal.HLogKey;
79 import org.apache.hadoop.hbase.wal.WALKey;
80 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
81 import org.apache.hadoop.hbase.util.Bytes;
82 import org.apache.hadoop.hbase.util.CoprocessorClassLoader;
83 import org.apache.hadoop.hbase.util.Pair;
84
85 import com.google.common.collect.ImmutableList;
86 import com.google.common.collect.Lists;
87 import com.google.protobuf.Message;
88 import com.google.protobuf.Service;
89
90
91
92
93
94 @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.COPROC)
95 @InterfaceStability.Evolving
96 public class RegionCoprocessorHost
97 extends CoprocessorHost<RegionCoprocessorHost.RegionEnvironment> {
98
99 private static final Log LOG = LogFactory.getLog(RegionCoprocessorHost.class);
100
101 private static ReferenceMap sharedDataMap =
102 new ReferenceMap(AbstractReferenceMap.HARD, AbstractReferenceMap.WEAK);
103
104
105 private final boolean hasCustomPostScannerFilterRow;
106
107
108
109
110
111 static class RegionEnvironment extends CoprocessorHost.Environment
112 implements RegionCoprocessorEnvironment {
113
114 private Region region;
115 private RegionServerServices rsServices;
116 ConcurrentMap<String, Object> sharedData;
117 private final boolean useLegacyPre;
118 private final boolean useLegacyPost;
119 private final MetricRegistry metricRegistry;
120
121
122
123
124
125
126 public RegionEnvironment(final Coprocessor impl, final int priority,
127 final int seq, final Configuration conf, final Region region,
128 final RegionServerServices services, final ConcurrentMap<String, Object> sharedData) {
129 super(impl, priority, seq, conf);
130 this.region = region;
131 this.rsServices = services;
132 this.sharedData = sharedData;
133
134
135
136
137 useLegacyPre = useLegacyMethod(impl.getClass(), "preWALRestore", ObserverContext.class,
138 HRegionInfo.class, WALKey.class, WALEdit.class);
139 useLegacyPost = useLegacyMethod(impl.getClass(), "postWALRestore", ObserverContext.class,
140 HRegionInfo.class, WALKey.class, WALEdit.class);
141 this.metricRegistry =
142 MetricsCoprocessor.createRegistryForRegionCoprocessor(impl.getClass().getName());
143 }
144
145
146 @Override
147 public Region getRegion() {
148 return region;
149 }
150
151
152 @Override
153 public RegionServerServices getRegionServerServices() {
154 return rsServices;
155 }
156
157 @Override
158 public void shutdown() {
159 super.shutdown();
160 MetricsCoprocessor.removeRegistry(this.metricRegistry);
161 }
162
163 @Override
164 public ConcurrentMap<String, Object> getSharedData() {
165 return sharedData;
166 }
167
168 @Override
169 public HRegionInfo getRegionInfo() {
170 return region.getRegionInfo();
171 }
172
173 @Override
174 public MetricRegistry getMetricRegistryForRegionServer() {
175 return metricRegistry;
176 }
177 }
178
179 static class TableCoprocessorAttribute {
180 private Path path;
181 private String className;
182 private int priority;
183 private Configuration conf;
184
185 public TableCoprocessorAttribute(Path path, String className, int priority,
186 Configuration conf) {
187 this.path = path;
188 this.className = className;
189 this.priority = priority;
190 this.conf = conf;
191 }
192
193 public Path getPath() {
194 return path;
195 }
196
197 public String getClassName() {
198 return className;
199 }
200
201 public int getPriority() {
202 return priority;
203 }
204
205 public Configuration getConf() {
206 return conf;
207 }
208 }
209
210
211 RegionServerServices rsServices;
212
213 Region region;
214
215
216
217
218
219
220
221 public RegionCoprocessorHost(final Region region,
222 final RegionServerServices rsServices, final Configuration conf) {
223 super(rsServices);
224 this.conf = conf;
225 this.rsServices = rsServices;
226 this.region = region;
227 this.pathPrefix = Integer.toString(this.region.getRegionInfo().hashCode());
228
229
230 loadSystemCoprocessors(conf, REGION_COPROCESSOR_CONF_KEY);
231
232
233 if (!region.getRegionInfo().getTable().isSystemTable()) {
234 loadSystemCoprocessors(conf, USER_REGION_COPROCESSOR_CONF_KEY);
235 }
236
237
238 loadTableCoprocessors(conf);
239
240
241 boolean hasCustomPostScannerFilterRow = false;
242 out: for (RegionEnvironment env: coprocessors) {
243 if (env.getInstance() instanceof RegionObserver) {
244 Class<?> clazz = env.getInstance().getClass();
245 for(;;) {
246 if (clazz == null) {
247
248 hasCustomPostScannerFilterRow = true;
249 break out;
250 }
251 if (clazz == BaseRegionObserver.class) {
252
253 break;
254 }
255 try {
256 clazz.getDeclaredMethod("postScannerFilterRow", ObserverContext.class,
257 InternalScanner.class, byte[].class, int.class, short.class, boolean.class);
258
259 hasCustomPostScannerFilterRow = true;
260 break out;
261 } catch (NoSuchMethodException ignore) {
262 }
263 clazz = clazz.getSuperclass();
264 }
265 }
266 }
267 this.hasCustomPostScannerFilterRow = hasCustomPostScannerFilterRow;
268 }
269
270 static List<TableCoprocessorAttribute> getTableCoprocessorAttrsFromSchema(Configuration conf,
271 HTableDescriptor htd) {
272 List<TableCoprocessorAttribute> result = Lists.newArrayList();
273 for (Map.Entry<ImmutableBytesWritable, ImmutableBytesWritable> e: htd.getValues().entrySet()) {
274 String key = Bytes.toString(e.getKey().get()).trim();
275 if (HConstants.CP_HTD_ATTR_KEY_PATTERN.matcher(key).matches()) {
276 String spec = Bytes.toString(e.getValue().get()).trim();
277
278 try {
279 Matcher matcher = HConstants.CP_HTD_ATTR_VALUE_PATTERN.matcher(spec);
280 if (matcher.matches()) {
281
282
283 Path path = matcher.group(1).trim().isEmpty() ?
284 null : new Path(matcher.group(1).trim());
285 String className = matcher.group(2).trim();
286 if (className.isEmpty()) {
287 LOG.error("Malformed table coprocessor specification: key=" +
288 key + ", spec: " + spec);
289 continue;
290 }
291 int priority = matcher.group(3).trim().isEmpty() ?
292 Coprocessor.PRIORITY_USER : Integer.parseInt(matcher.group(3));
293 String cfgSpec = null;
294 try {
295 cfgSpec = matcher.group(4);
296 } catch (IndexOutOfBoundsException ex) {
297
298 }
299 Configuration ourConf;
300 if (cfgSpec != null && !cfgSpec.trim().equals("|")) {
301 cfgSpec = cfgSpec.substring(cfgSpec.indexOf('|') + 1);
302
303 ourConf = new Configuration(false);
304 HBaseConfiguration.merge(ourConf, conf);
305 Matcher m = HConstants.CP_HTD_ATTR_VALUE_PARAM_PATTERN.matcher(cfgSpec);
306 while (m.find()) {
307 ourConf.set(m.group(1), m.group(2));
308 }
309 } else {
310 ourConf = conf;
311 }
312 result.add(new TableCoprocessorAttribute(path, className, priority, ourConf));
313 } else {
314 LOG.error("Malformed table coprocessor specification: key=" + key +
315 ", spec: " + spec);
316 }
317 } catch (Exception ioe) {
318 LOG.error("Malformed table coprocessor specification: key=" + key +
319 ", spec: " + spec);
320 }
321 }
322 }
323 return result;
324 }
325
326
327
328
329
330
331
332
333 public static void testTableCoprocessorAttrs(final Configuration conf,
334 final HTableDescriptor htd) throws IOException {
335 String pathPrefix = UUID.randomUUID().toString();
336 for (TableCoprocessorAttribute attr: getTableCoprocessorAttrsFromSchema(conf, htd)) {
337 if (attr.getPriority() < 0) {
338 throw new IOException("Priority for coprocessor " + attr.getClassName() +
339 " cannot be less than 0");
340 }
341 ClassLoader old = Thread.currentThread().getContextClassLoader();
342 try {
343 ClassLoader cl;
344 if (attr.getPath() != null) {
345 cl = CoprocessorClassLoader.getClassLoader(attr.getPath(),
346 CoprocessorHost.class.getClassLoader(), pathPrefix, conf);
347 } else {
348 cl = CoprocessorHost.class.getClassLoader();
349 }
350 Thread.currentThread().setContextClassLoader(cl);
351 cl.loadClass(attr.getClassName());
352 } catch (ClassNotFoundException e) {
353 throw new IOException("Class " + attr.getClassName() + " cannot be loaded", e);
354 } finally {
355 Thread.currentThread().setContextClassLoader(old);
356 }
357 }
358 }
359
360 void loadTableCoprocessors(final Configuration conf) {
361 boolean coprocessorsEnabled = conf.getBoolean(COPROCESSORS_ENABLED_CONF_KEY,
362 DEFAULT_COPROCESSORS_ENABLED);
363 boolean tableCoprocessorsEnabled = conf.getBoolean(USER_COPROCESSORS_ENABLED_CONF_KEY,
364 DEFAULT_USER_COPROCESSORS_ENABLED);
365 if (!(coprocessorsEnabled && tableCoprocessorsEnabled)) {
366 return;
367 }
368
369
370
371 List<RegionEnvironment> configured = new ArrayList<RegionEnvironment>();
372 for (TableCoprocessorAttribute attr: getTableCoprocessorAttrsFromSchema(conf,
373 region.getTableDesc())) {
374
375 try {
376 RegionEnvironment env = load(attr.getPath(), attr.getClassName(), attr.getPriority(),
377 attr.getConf());
378 configured.add(env);
379 LOG.info("Loaded coprocessor " + attr.getClassName() + " from HTD of " +
380 region.getTableDesc().getTableName().getNameAsString() + " successfully.");
381 } catch (Throwable t) {
382
383 if (conf.getBoolean(ABORT_ON_ERROR_KEY, DEFAULT_ABORT_ON_ERROR)) {
384 abortServer(attr.getClassName(), t);
385 } else {
386 LOG.error("Failed to load coprocessor " + attr.getClassName(), t);
387 }
388 }
389 }
390
391 coprocessors.addAll(configured);
392 }
393
394 @Override
395 public RegionEnvironment createEnvironment(Class<?> implClass,
396 Coprocessor instance, int priority, int seq, Configuration conf) {
397
398
399
400
401
402 for (Object itf : ClassUtils.getAllInterfaces(implClass)) {
403 Class<?> c = (Class<?>) itf;
404 if (CoprocessorService.class.isAssignableFrom(c)) {
405 region.registerService( ((CoprocessorService)instance).getService() );
406 }
407 }
408 ConcurrentMap<String, Object> classData;
409
410 synchronized (sharedDataMap) {
411
412
413 classData = (ConcurrentMap<String, Object>)sharedDataMap.get(implClass.getName());
414 if (classData == null) {
415 classData = new ConcurrentHashMap<String, Object>();
416 sharedDataMap.put(implClass.getName(), classData);
417 }
418 }
419 return new RegionEnvironment(instance, priority, seq, conf, region,
420 rsServices, classData);
421 }
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436 private void handleCoprocessorThrowableNoRethrow(
437 final CoprocessorEnvironment env, final Throwable e) {
438 try {
439 handleCoprocessorThrowable(env,e);
440 } catch (IOException ioe) {
441
442 LOG.warn(
443 "handleCoprocessorThrowable() threw an IOException while attempting to handle Throwable " +
444 e + ". Ignoring.",e);
445 }
446 }
447
448
449
450
451
452
453 public void preOpen() throws IOException {
454 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
455 @Override
456 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
457 throws IOException {
458 oserver.preOpen(ctx);
459 }
460 });
461 }
462
463
464
465
466 public void postOpen() {
467 try {
468 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
469 @Override
470 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
471 throws IOException {
472 oserver.postOpen(ctx);
473 }
474 });
475 } catch (IOException e) {
476 LOG.warn(e);
477 }
478 }
479
480
481
482
483 public void postLogReplay() {
484 try {
485 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
486 @Override
487 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
488 throws IOException {
489 oserver.postLogReplay(ctx);
490 }
491 });
492 } catch (IOException e) {
493 LOG.warn(e);
494 }
495 }
496
497
498
499
500
501 public void preClose(final boolean abortRequested) throws IOException {
502 execOperation(false, new RegionOperation() {
503 @Override
504 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
505 throws IOException {
506 oserver.preClose(ctx, abortRequested);
507 }
508 });
509 }
510
511
512
513
514
515 public void postClose(final boolean abortRequested) {
516 try {
517 execOperation(false, new RegionOperation() {
518 @Override
519 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
520 throws IOException {
521 oserver.postClose(ctx, abortRequested);
522 }
523 @Override
524 public void postEnvCall(RegionEnvironment env) {
525 shutdown(env);
526 }
527 });
528 } catch (IOException e) {
529 LOG.warn(e);
530 }
531 }
532
533
534
535
536
537
538 public InternalScanner preCompactScannerOpen(final Store store,
539 final List<StoreFileScanner> scanners, final ScanType scanType, final long earliestPutTs,
540 final CompactionRequest request, final long readPoint) throws IOException {
541 return execOperationWithResult(null,
542 coprocessors.isEmpty() ? null : new RegionOperationWithResult<InternalScanner>() {
543 @Override
544 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
545 throws IOException {
546 setResult(oserver.preCompactScannerOpen(ctx, store, scanners, scanType,
547 earliestPutTs, getResult(), request, readPoint));
548 }
549 });
550 }
551
552
553
554
555
556
557
558
559
560
561 public boolean preCompactSelection(final Store store, final List<StoreFile> candidates,
562 final CompactionRequest request) throws IOException {
563 return execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
564 @Override
565 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
566 throws IOException {
567 oserver.preCompactSelection(ctx, store, candidates, request);
568 }
569 });
570 }
571
572
573
574
575
576
577
578
579 public void postCompactSelection(final Store store, final ImmutableList<StoreFile> selected,
580 final CompactionRequest request) {
581 try {
582 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
583 @Override
584 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
585 throws IOException {
586 oserver.postCompactSelection(ctx, store, selected, request);
587 }
588 });
589 } catch (IOException e) {
590 LOG.warn(e);
591 }
592 }
593
594
595
596
597
598
599
600
601
602 public InternalScanner preCompact(final Store store, final InternalScanner scanner,
603 final ScanType scanType, final CompactionRequest request) throws IOException {
604 return execOperationWithResult(false, scanner,
605 coprocessors.isEmpty() ? null : new RegionOperationWithResult<InternalScanner>() {
606 @Override
607 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
608 throws IOException {
609 setResult(oserver.preCompact(ctx, store, getResult(), scanType, request));
610 }
611 });
612 }
613
614
615
616
617
618
619
620
621 public void postCompact(final Store store, final StoreFile resultFile,
622 final CompactionRequest request) throws IOException {
623 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
624 @Override
625 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
626 throws IOException {
627 oserver.postCompact(ctx, store, resultFile, request);
628 }
629 });
630 }
631
632
633
634
635
636 public InternalScanner preFlush(final Store store, final InternalScanner scanner)
637 throws IOException {
638 return execOperationWithResult(false, scanner,
639 coprocessors.isEmpty() ? null : new RegionOperationWithResult<InternalScanner>() {
640 @Override
641 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
642 throws IOException {
643 setResult(oserver.preFlush(ctx, store, getResult()));
644 }
645 });
646 }
647
648
649
650
651
652 public void preFlush() throws IOException {
653 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
654 @Override
655 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
656 throws IOException {
657 oserver.preFlush(ctx);
658 }
659 });
660 }
661
662
663
664
665
666
667 public InternalScanner preFlushScannerOpen(final Store store,
668 final KeyValueScanner memstoreScanner, final long readPoint) throws IOException {
669 return execOperationWithResult(null,
670 coprocessors.isEmpty() ? null : new RegionOperationWithResult<InternalScanner>() {
671 @Override
672 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
673 throws IOException {
674 setResult(oserver.preFlushScannerOpen(ctx, store, memstoreScanner, getResult(), readPoint));
675 }
676 });
677 }
678
679
680
681
682
683 public void postFlush() throws IOException {
684 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
685 @Override
686 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
687 throws IOException {
688 oserver.postFlush(ctx);
689 }
690 });
691 }
692
693
694
695
696
697 public void postFlush(final Store store, final StoreFile storeFile) throws IOException {
698 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
699 @Override
700 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
701 throws IOException {
702 oserver.postFlush(ctx, store, storeFile);
703 }
704 });
705 }
706
707
708
709
710
711
712 public void preSplit() throws IOException {
713 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
714 @Override
715 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
716 throws IOException {
717 oserver.preSplit(ctx);
718 }
719 });
720 }
721
722
723
724
725
726 public void preSplit(final byte[] splitRow) throws IOException {
727 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
728 @Override
729 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
730 throws IOException {
731 oserver.preSplit(ctx, splitRow);
732 }
733 });
734 }
735
736
737
738
739
740
741
742 public void postSplit(final Region l, final Region r) throws IOException {
743 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
744 @Override
745 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
746 throws IOException {
747 oserver.postSplit(ctx, l, r);
748 }
749 });
750 }
751
752 public boolean preSplitBeforePONR(final byte[] splitKey,
753 final List<Mutation> metaEntries) throws IOException {
754 return execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
755 @Override
756 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
757 throws IOException {
758 oserver.preSplitBeforePONR(ctx, splitKey, metaEntries);
759 }
760 });
761 }
762
763 public void preSplitAfterPONR() throws IOException {
764 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
765 @Override
766 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
767 throws IOException {
768 oserver.preSplitAfterPONR(ctx);
769 }
770 });
771 }
772
773
774
775
776
777 public void preRollBackSplit() throws IOException {
778 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
779 @Override
780 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
781 throws IOException {
782 oserver.preRollBackSplit(ctx);
783 }
784 });
785 }
786
787
788
789
790
791 public void postRollBackSplit() throws IOException {
792 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
793 @Override
794 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
795 throws IOException {
796 oserver.postRollBackSplit(ctx);
797 }
798 });
799 }
800
801
802
803
804
805 public void postCompleteSplit() throws IOException {
806 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
807 @Override
808 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
809 throws IOException {
810 oserver.postCompleteSplit(ctx);
811 }
812 });
813 }
814
815
816
817
818
819
820
821
822
823
824 public boolean preGetClosestRowBefore(final byte[] row, final byte[] family,
825 final Result result) throws IOException {
826 return execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
827 @Override
828 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
829 throws IOException {
830 oserver.preGetClosestRowBefore(ctx, row, family, result);
831 }
832 });
833 }
834
835
836
837
838
839
840
841 public void postGetClosestRowBefore(final byte[] row, final byte[] family,
842 final Result result) throws IOException {
843 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
844 @Override
845 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
846 throws IOException {
847 oserver.postGetClosestRowBefore(ctx, row, family, result);
848 }
849 });
850 }
851
852
853
854
855
856
857 public boolean preGet(final Get get, final List<Cell> results)
858 throws IOException {
859 return execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
860 @Override
861 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
862 throws IOException {
863 oserver.preGetOp(ctx, get, results);
864 }
865 });
866 }
867
868
869
870
871
872
873 public void postGet(final Get get, final List<Cell> results)
874 throws IOException {
875 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
876 @Override
877 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
878 throws IOException {
879 oserver.postGetOp(ctx, get, results);
880 }
881 });
882 }
883
884
885
886
887
888
889
890 public Boolean preExists(final Get get) throws IOException {
891 return execOperationWithResult(true, false,
892 coprocessors.isEmpty() ? null : new RegionOperationWithResult<Boolean>() {
893 @Override
894 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
895 throws IOException {
896 setResult(oserver.preExists(ctx, get, getResult()));
897 }
898 });
899 }
900
901
902
903
904
905
906
907 public boolean postExists(final Get get, boolean exists)
908 throws IOException {
909 return execOperationWithResult(exists,
910 coprocessors.isEmpty() ? null : new RegionOperationWithResult<Boolean>() {
911 @Override
912 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
913 throws IOException {
914 setResult(oserver.postExists(ctx, get, getResult()));
915 }
916 });
917 }
918
919
920
921
922
923
924
925
926 public boolean prePut(final Put put, final WALEdit edit, final Durability durability)
927 throws IOException {
928 return execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
929 @Override
930 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
931 throws IOException {
932 oserver.prePut(ctx, put, edit, durability);
933 }
934 });
935 }
936
937
938
939
940
941
942
943
944
945
946
947 public boolean prePrepareTimeStampForDeleteVersion(final Mutation mutation,
948 final Cell kv, final byte[] byteNow, final Get get) throws IOException {
949 return execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
950 @Override
951 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
952 throws IOException {
953 oserver.prePrepareTimeStampForDeleteVersion(ctx, mutation, kv, byteNow, get);
954 }
955 });
956 }
957
958
959
960
961
962
963
964 public void postPut(final Put put, final WALEdit edit, final Durability durability)
965 throws IOException {
966 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
967 @Override
968 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
969 throws IOException {
970 oserver.postPut(ctx, put, edit, durability);
971 }
972 });
973 }
974
975
976
977
978
979
980
981
982 public boolean preDelete(final Delete delete, final WALEdit edit, final Durability durability)
983 throws IOException {
984 return execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
985 @Override
986 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
987 throws IOException {
988 oserver.preDelete(ctx, delete, edit, durability);
989 }
990 });
991 }
992
993
994
995
996
997
998
999 public void postDelete(final Delete delete, final WALEdit edit, final Durability durability)
1000 throws IOException {
1001 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
1002 @Override
1003 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1004 throws IOException {
1005 oserver.postDelete(ctx, delete, edit, durability);
1006 }
1007 });
1008 }
1009
1010
1011
1012
1013
1014
1015 public boolean preBatchMutate(
1016 final MiniBatchOperationInProgress<Mutation> miniBatchOp) throws IOException {
1017 return execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
1018 @Override
1019 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1020 throws IOException {
1021 oserver.preBatchMutate(ctx, miniBatchOp);
1022 }
1023 });
1024 }
1025
1026
1027
1028
1029
1030 public void postBatchMutate(
1031 final MiniBatchOperationInProgress<Mutation> miniBatchOp) throws IOException {
1032 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
1033 @Override
1034 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1035 throws IOException {
1036 oserver.postBatchMutate(ctx, miniBatchOp);
1037 }
1038 });
1039 }
1040
1041 public void postBatchMutateIndispensably(
1042 final MiniBatchOperationInProgress<Mutation> miniBatchOp, final boolean success)
1043 throws IOException {
1044 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
1045 @Override
1046 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1047 throws IOException {
1048 oserver.postBatchMutateIndispensably(ctx, miniBatchOp, success);
1049 }
1050 });
1051 }
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064 public Boolean preCheckAndPut(final byte [] row, final byte [] family,
1065 final byte [] qualifier, final CompareOp compareOp,
1066 final ByteArrayComparable comparator, final Put put)
1067 throws IOException {
1068 return execOperationWithResult(true, false,
1069 coprocessors.isEmpty() ? null : new RegionOperationWithResult<Boolean>() {
1070 @Override
1071 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1072 throws IOException {
1073 setResult(oserver.preCheckAndPut(ctx, row, family, qualifier,
1074 compareOp, comparator, put, getResult()));
1075 }
1076 });
1077 }
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090 public Boolean preCheckAndPutAfterRowLock(final byte[] row, final byte[] family,
1091 final byte[] qualifier, final CompareOp compareOp, final ByteArrayComparable comparator,
1092 final Put put) throws IOException {
1093 return execOperationWithResult(true, false,
1094 coprocessors.isEmpty() ? null : new RegionOperationWithResult<Boolean>() {
1095 @Override
1096 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1097 throws IOException {
1098 setResult(oserver.preCheckAndPutAfterRowLock(ctx, row, family, qualifier,
1099 compareOp, comparator, put, getResult()));
1100 }
1101 });
1102 }
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113 public boolean postCheckAndPut(final byte [] row, final byte [] family,
1114 final byte [] qualifier, final CompareOp compareOp,
1115 final ByteArrayComparable comparator, final Put put,
1116 boolean result) throws IOException {
1117 return execOperationWithResult(result,
1118 coprocessors.isEmpty() ? null : new RegionOperationWithResult<Boolean>() {
1119 @Override
1120 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1121 throws IOException {
1122 setResult(oserver.postCheckAndPut(ctx, row, family, qualifier,
1123 compareOp, comparator, put, getResult()));
1124 }
1125 });
1126 }
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139 public Boolean preCheckAndDelete(final byte [] row, final byte [] family,
1140 final byte [] qualifier, final CompareOp compareOp,
1141 final ByteArrayComparable comparator, final Delete delete)
1142 throws IOException {
1143 return execOperationWithResult(true, false,
1144 coprocessors.isEmpty() ? null : new RegionOperationWithResult<Boolean>() {
1145 @Override
1146 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1147 throws IOException {
1148 setResult(oserver.preCheckAndDelete(ctx, row, family,
1149 qualifier, compareOp, comparator, delete, getResult()));
1150 }
1151 });
1152 }
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165 public Boolean preCheckAndDeleteAfterRowLock(final byte[] row, final byte[] family,
1166 final byte[] qualifier, final CompareOp compareOp, final ByteArrayComparable comparator,
1167 final Delete delete) throws IOException {
1168 return execOperationWithResult(true, false,
1169 coprocessors.isEmpty() ? null : new RegionOperationWithResult<Boolean>() {
1170 @Override
1171 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1172 throws IOException {
1173 setResult(oserver.preCheckAndDeleteAfterRowLock(ctx, row,
1174 family, qualifier, compareOp, comparator, delete, getResult()));
1175 }
1176 });
1177 }
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188 public boolean postCheckAndDelete(final byte [] row, final byte [] family,
1189 final byte [] qualifier, final CompareOp compareOp,
1190 final ByteArrayComparable comparator, final Delete delete,
1191 boolean result) throws IOException {
1192 return execOperationWithResult(result,
1193 coprocessors.isEmpty() ? null : new RegionOperationWithResult<Boolean>() {
1194 @Override
1195 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1196 throws IOException {
1197 setResult(oserver.postCheckAndDelete(ctx, row, family,
1198 qualifier, compareOp, comparator, delete, getResult()));
1199 }
1200 });
1201 }
1202
1203
1204
1205
1206
1207
1208
1209 public Result preAppend(final Append append) throws IOException {
1210 return execOperationWithResult(true, null,
1211 coprocessors.isEmpty() ? null : new RegionOperationWithResult<Result>() {
1212 @Override
1213 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1214 throws IOException {
1215 setResult(oserver.preAppend(ctx, append));
1216 }
1217 });
1218 }
1219
1220
1221
1222
1223
1224
1225
1226 public Result preAppendAfterRowLock(final Append append) throws IOException {
1227 return execOperationWithResult(true, null,
1228 coprocessors.isEmpty() ? null : new RegionOperationWithResult<Result>() {
1229 @Override
1230 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1231 throws IOException {
1232 setResult(oserver.preAppendAfterRowLock(ctx, append));
1233 }
1234 });
1235 }
1236
1237
1238
1239
1240
1241
1242
1243 public Result preIncrement(final Increment increment) throws IOException {
1244 return execOperationWithResult(true, null,
1245 coprocessors.isEmpty() ? null : new RegionOperationWithResult<Result>() {
1246 @Override
1247 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1248 throws IOException {
1249 setResult(oserver.preIncrement(ctx, increment));
1250 }
1251 });
1252 }
1253
1254
1255
1256
1257
1258
1259
1260 public Result preIncrementAfterRowLock(final Increment increment) throws IOException {
1261 return execOperationWithResult(true, null,
1262 coprocessors.isEmpty() ? null : new RegionOperationWithResult<Result>() {
1263 @Override
1264 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1265 throws IOException {
1266 setResult(oserver.preIncrementAfterRowLock(ctx, increment));
1267 }
1268 });
1269 }
1270
1271
1272
1273
1274
1275
1276 public Result postAppend(final Append append, final Result result) throws IOException {
1277 return execOperationWithResult(result,
1278 coprocessors.isEmpty() ? null : new RegionOperationWithResult<Result>() {
1279 @Override
1280 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1281 throws IOException {
1282 setResult(oserver.postAppend(ctx, append, result));
1283 }
1284 });
1285 }
1286
1287
1288
1289
1290
1291
1292 public Result postIncrement(final Increment increment, Result result) throws IOException {
1293 return execOperationWithResult(result,
1294 coprocessors.isEmpty() ? null : new RegionOperationWithResult<Result>() {
1295 @Override
1296 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1297 throws IOException {
1298 setResult(oserver.postIncrement(ctx, increment, getResult()));
1299 }
1300 });
1301 }
1302
1303
1304
1305
1306
1307
1308
1309 public RegionScanner preScannerOpen(final Scan scan) throws IOException {
1310 return execOperationWithResult(true, null,
1311 coprocessors.isEmpty() ? null : new RegionOperationWithResult<RegionScanner>() {
1312 @Override
1313 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1314 throws IOException {
1315 setResult(oserver.preScannerOpen(ctx, scan, getResult()));
1316 }
1317 });
1318 }
1319
1320
1321
1322
1323
1324
1325 public KeyValueScanner preStoreScannerOpen(final Store store, final Scan scan,
1326 final NavigableSet<byte[]> targetCols) throws IOException {
1327 return execOperationWithResult(null,
1328 coprocessors.isEmpty() ? null : new RegionOperationWithResult<KeyValueScanner>() {
1329 @Override
1330 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1331 throws IOException {
1332 setResult(oserver.preStoreScannerOpen(ctx, store, scan, targetCols, getResult()));
1333 }
1334 });
1335 }
1336
1337
1338
1339
1340
1341
1342
1343 public RegionScanner postScannerOpen(final Scan scan, RegionScanner s) throws IOException {
1344 return execOperationWithResult(s,
1345 coprocessors.isEmpty() ? null : new RegionOperationWithResult<RegionScanner>() {
1346 @Override
1347 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1348 throws IOException {
1349 setResult(oserver.postScannerOpen(ctx, scan, getResult()));
1350 }
1351 });
1352 }
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362 public Boolean preScannerNext(final InternalScanner s,
1363 final List<Result> results, final int limit) throws IOException {
1364 return execOperationWithResult(true, false,
1365 coprocessors.isEmpty() ? null : new RegionOperationWithResult<Boolean>() {
1366 @Override
1367 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1368 throws IOException {
1369 setResult(oserver.preScannerNext(ctx, s, results, limit, getResult()));
1370 }
1371 });
1372 }
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382 public boolean postScannerNext(final InternalScanner s,
1383 final List<Result> results, final int limit, boolean hasMore)
1384 throws IOException {
1385 return execOperationWithResult(hasMore,
1386 coprocessors.isEmpty() ? null : new RegionOperationWithResult<Boolean>() {
1387 @Override
1388 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1389 throws IOException {
1390 setResult(oserver.postScannerNext(ctx, s, results, limit, getResult()));
1391 }
1392 });
1393 }
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405 public boolean postScannerFilterRow(final InternalScanner s, final byte[] currentRow,
1406 final int offset, final short length) throws IOException {
1407
1408 if (!hasCustomPostScannerFilterRow) return true;
1409 return execOperationWithResult(true,
1410 coprocessors.isEmpty() ? null : new RegionOperationWithResult<Boolean>() {
1411 @Override
1412 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1413 throws IOException {
1414 setResult(oserver.postScannerFilterRow(ctx, s, currentRow, offset,length, getResult()));
1415 }
1416 });
1417 }
1418
1419
1420
1421
1422
1423
1424 public boolean preScannerClose(final InternalScanner s) throws IOException {
1425 return execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
1426 @Override
1427 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1428 throws IOException {
1429 oserver.preScannerClose(ctx, s);
1430 }
1431 });
1432 }
1433
1434
1435
1436
1437 public void postScannerClose(final InternalScanner s) throws IOException {
1438 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
1439 @Override
1440 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1441 throws IOException {
1442 oserver.postScannerClose(ctx, s);
1443 }
1444 });
1445 }
1446
1447
1448
1449
1450
1451
1452
1453
1454 public boolean preWALRestore(final HRegionInfo info, final WALKey logKey,
1455 final WALEdit logEdit) throws IOException {
1456 return execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
1457 @Override
1458 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1459 throws IOException {
1460
1461
1462 final RegionEnvironment env = (RegionEnvironment)ctx.getEnvironment();
1463 if (env.useLegacyPre) {
1464 if (logKey instanceof HLogKey) {
1465 oserver.preWALRestore(ctx, info, (HLogKey)logKey, logEdit);
1466 } else {
1467 legacyWarning(oserver.getClass(), "There are wal keys present that are not HLogKey.");
1468 }
1469 } else {
1470 oserver.preWALRestore(ctx, info, logKey, logEdit);
1471 }
1472 }
1473 });
1474 }
1475
1476
1477
1478
1479
1480 @Deprecated
1481 public boolean preWALRestore(final HRegionInfo info, final HLogKey logKey,
1482 final WALEdit logEdit) throws IOException {
1483 return preWALRestore(info, (WALKey)logKey, logEdit);
1484 }
1485
1486
1487
1488
1489
1490
1491
1492 public void postWALRestore(final HRegionInfo info, final WALKey logKey, final WALEdit logEdit)
1493 throws IOException {
1494 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
1495 @Override
1496 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1497 throws IOException {
1498
1499
1500 final RegionEnvironment env = (RegionEnvironment)ctx.getEnvironment();
1501 if (env.useLegacyPost) {
1502 if (logKey instanceof HLogKey) {
1503 oserver.postWALRestore(ctx, info, (HLogKey)logKey, logEdit);
1504 } else {
1505 legacyWarning(oserver.getClass(), "There are wal keys present that are not HLogKey.");
1506 }
1507 } else {
1508 oserver.postWALRestore(ctx, info, logKey, logEdit);
1509 }
1510 }
1511 });
1512 }
1513
1514
1515
1516
1517 @Deprecated
1518 public void postWALRestore(final HRegionInfo info, final HLogKey logKey, final WALEdit logEdit)
1519 throws IOException {
1520 postWALRestore(info, (WALKey)logKey, logEdit);
1521 }
1522
1523 public boolean preCommitStoreFile(final byte[] family, final List<Pair<Path, Path>> pairs)
1524 throws IOException {
1525 return execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
1526 @Override
1527 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1528 throws IOException {
1529 oserver.preCommitStoreFile(ctx, family, pairs);
1530 }
1531 });
1532 }
1533 public void postCommitStoreFile(final byte[] family, final Path srcPath, final Path dstPath)
1534 throws IOException {
1535 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
1536 @Override
1537 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1538 throws IOException {
1539 oserver.postCommitStoreFile(ctx, family, srcPath, dstPath);
1540 }
1541 });
1542 }
1543
1544
1545
1546
1547
1548
1549 public boolean preBulkLoadHFile(final List<Pair<byte[], String>> familyPaths) throws IOException {
1550 return execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
1551 @Override
1552 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1553 throws IOException {
1554 oserver.preBulkLoadHFile(ctx, familyPaths);
1555 }
1556 });
1557 }
1558
1559
1560
1561
1562
1563
1564
1565 public boolean postBulkLoadHFile(final List<Pair<byte[], String>> familyPaths,
1566 boolean hasLoaded) throws IOException {
1567 return execOperationWithResult(hasLoaded,
1568 coprocessors.isEmpty() ? null : new RegionOperationWithResult<Boolean>() {
1569 @Override
1570 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1571 throws IOException {
1572 setResult(oserver.postBulkLoadHFile(ctx, familyPaths, getResult()));
1573 }
1574 });
1575 }
1576
1577 public void postStartRegionOperation(final Operation op) throws IOException {
1578 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
1579 @Override
1580 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1581 throws IOException {
1582 oserver.postStartRegionOperation(ctx, op);
1583 }
1584 });
1585 }
1586
1587 public void postCloseRegionOperation(final Operation op) throws IOException {
1588 execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
1589 @Override
1590 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1591 throws IOException {
1592 oserver.postCloseRegionOperation(ctx, op);
1593 }
1594 });
1595 }
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608 public StoreFile.Reader preStoreFileReaderOpen(final FileSystem fs, final Path p,
1609 final FSDataInputStreamWrapper in, final long size, final CacheConfig cacheConf,
1610 final Reference r) throws IOException {
1611 return execOperationWithResult(null,
1612 coprocessors.isEmpty() ? null : new RegionOperationWithResult<StoreFile.Reader>() {
1613 @Override
1614 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1615 throws IOException {
1616 setResult(oserver.preStoreFileReaderOpen(ctx, fs, p, in, size, cacheConf, r, getResult()));
1617 }
1618 });
1619 }
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632 public StoreFile.Reader postStoreFileReaderOpen(final FileSystem fs, final Path p,
1633 final FSDataInputStreamWrapper in, final long size, final CacheConfig cacheConf,
1634 final Reference r, final StoreFile.Reader reader) throws IOException {
1635 return execOperationWithResult(reader,
1636 coprocessors.isEmpty() ? null : new RegionOperationWithResult<StoreFile.Reader>() {
1637 @Override
1638 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1639 throws IOException {
1640 setResult(oserver.postStoreFileReaderOpen(ctx, fs, p, in, size, cacheConf, r, getResult()));
1641 }
1642 });
1643 }
1644
1645 public Cell postMutationBeforeWAL(final MutationType opType, final Mutation mutation,
1646 final Cell oldCell, Cell newCell) throws IOException {
1647 return execOperationWithResult(newCell,
1648 coprocessors.isEmpty() ? null : new RegionOperationWithResult<Cell>() {
1649 @Override
1650 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1651 throws IOException {
1652 setResult(oserver.postMutationBeforeWAL(ctx, opType, mutation, oldCell, getResult()));
1653 }
1654 });
1655 }
1656
1657 public Message preEndpointInvocation(final Service service, final String methodName,
1658 Message request) throws IOException {
1659 return execOperationWithResult(request,
1660 coprocessors.isEmpty() ? null : new EndpointOperationWithResult<Message>() {
1661 @Override
1662 public void call(EndpointObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1663 throws IOException {
1664 setResult(oserver.preEndpointInvocation(ctx, service, methodName, getResult()));
1665 }
1666 });
1667 }
1668
1669 public void postEndpointInvocation(final Service service, final String methodName,
1670 final Message request, final Message.Builder responseBuilder) throws IOException {
1671 execOperation(coprocessors.isEmpty() ? null : new EndpointOperation() {
1672 @Override
1673 public void call(EndpointObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1674 throws IOException {
1675 oserver.postEndpointInvocation(ctx, service, methodName, request, responseBuilder);
1676 }
1677 });
1678 }
1679
1680 public DeleteTracker postInstantiateDeleteTracker(DeleteTracker tracker) throws IOException {
1681 return execOperationWithResult(tracker,
1682 coprocessors.isEmpty() ? null : new RegionOperationWithResult<DeleteTracker>() {
1683 @Override
1684 public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
1685 throws IOException {
1686 setResult(oserver.postInstantiateDeleteTracker(ctx, getResult()));
1687 }
1688 });
1689 }
1690
1691 private static abstract class CoprocessorOperation
1692 extends ObserverContext<RegionCoprocessorEnvironment> {
1693 public abstract void call(Coprocessor observer,
1694 ObserverContext<RegionCoprocessorEnvironment> ctx) throws IOException;
1695 public abstract boolean hasCall(Coprocessor observer);
1696 public void postEnvCall(RegionEnvironment env) { }
1697 }
1698
1699 private static abstract class RegionOperation extends CoprocessorOperation {
1700 public abstract void call(RegionObserver observer,
1701 ObserverContext<RegionCoprocessorEnvironment> ctx) throws IOException;
1702
1703 @Override
1704 public boolean hasCall(Coprocessor observer) {
1705 return observer instanceof RegionObserver;
1706 }
1707
1708 @Override
1709 public void call(Coprocessor observer, ObserverContext<RegionCoprocessorEnvironment> ctx)
1710 throws IOException {
1711 call((RegionObserver)observer, ctx);
1712 }
1713 }
1714
1715 private static abstract class RegionOperationWithResult<T> extends RegionOperation {
1716 private T result = null;
1717 public void setResult(final T result) { this.result = result; }
1718 public T getResult() { return this.result; }
1719 }
1720
1721 private static abstract class EndpointOperation extends CoprocessorOperation {
1722 public abstract void call(EndpointObserver observer,
1723 ObserverContext<RegionCoprocessorEnvironment> ctx) throws IOException;
1724
1725 @Override
1726 public boolean hasCall(Coprocessor observer) {
1727 return observer instanceof EndpointObserver;
1728 }
1729
1730 @Override
1731 public void call(Coprocessor observer, ObserverContext<RegionCoprocessorEnvironment> ctx)
1732 throws IOException {
1733 call((EndpointObserver)observer, ctx);
1734 }
1735 }
1736
1737 private static abstract class EndpointOperationWithResult<T> extends EndpointOperation {
1738 private T result = null;
1739 public void setResult(final T result) { this.result = result; }
1740 public T getResult() { return this.result; }
1741 }
1742
1743 private boolean execOperation(final CoprocessorOperation ctx)
1744 throws IOException {
1745 return execOperation(true, ctx);
1746 }
1747
1748 private <T> T execOperationWithResult(final T defaultValue,
1749 final RegionOperationWithResult<T> ctx) throws IOException {
1750 if (ctx == null) return defaultValue;
1751 ctx.setResult(defaultValue);
1752 execOperation(true, ctx);
1753 return ctx.getResult();
1754 }
1755
1756 private <T> T execOperationWithResult(final boolean ifBypass, final T defaultValue,
1757 final RegionOperationWithResult<T> ctx) throws IOException {
1758 boolean bypass = false;
1759 T result = defaultValue;
1760 if (ctx != null) {
1761 ctx.setResult(defaultValue);
1762 bypass = execOperation(true, ctx);
1763 result = ctx.getResult();
1764 }
1765 return bypass == ifBypass ? result : null;
1766 }
1767
1768 private <T> T execOperationWithResult(final T defaultValue,
1769 final EndpointOperationWithResult<T> ctx) throws IOException {
1770 if (ctx == null) return defaultValue;
1771 ctx.setResult(defaultValue);
1772 execOperation(true, ctx);
1773 return ctx.getResult();
1774 }
1775
1776 private boolean execOperation(final boolean earlyExit, final CoprocessorOperation ctx)
1777 throws IOException {
1778 boolean bypass = false;
1779 List<RegionEnvironment> envs = coprocessors.get();
1780 for (int i = 0; i < envs.size(); i++) {
1781 RegionEnvironment env = envs.get(i);
1782 Coprocessor observer = env.getInstance();
1783 if (ctx.hasCall(observer)) {
1784 ctx.prepare(env);
1785 Thread currentThread = Thread.currentThread();
1786 ClassLoader cl = currentThread.getContextClassLoader();
1787 try {
1788 currentThread.setContextClassLoader(env.getClassLoader());
1789 ctx.call(observer, ctx);
1790 } catch (Throwable e) {
1791 handleCoprocessorThrowable(env, e);
1792 } finally {
1793 currentThread.setContextClassLoader(cl);
1794 }
1795 bypass |= ctx.shouldBypass();
1796 if (earlyExit && ctx.shouldComplete()) {
1797 break;
1798 }
1799 }
1800
1801 ctx.postEnvCall(env);
1802 }
1803 return bypass;
1804 }
1805 }