1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.replication.regionserver;
20
21 import com.google.common.collect.Lists;
22 import com.google.common.util.concurrent.ListenableFuture;
23 import com.google.common.util.concurrent.Service;
24
25 import java.io.IOException;
26 import java.util.ArrayList;
27 import java.util.Collection;
28 import java.util.Comparator;
29 import java.util.HashMap;
30 import java.util.List;
31 import java.util.Map;
32 import java.util.UUID;
33 import java.util.concurrent.ConcurrentHashMap;
34 import java.util.concurrent.PriorityBlockingQueue;
35 import java.util.concurrent.TimeUnit;
36 import java.util.concurrent.atomic.AtomicLong;
37
38 import org.apache.commons.lang.StringUtils;
39 import org.apache.commons.logging.Log;
40 import org.apache.commons.logging.LogFactory;
41 import org.apache.hadoop.conf.Configuration;
42 import org.apache.hadoop.fs.FileStatus;
43 import org.apache.hadoop.fs.FileSystem;
44 import org.apache.hadoop.fs.Path;
45 import org.apache.hadoop.hbase.Cell;
46 import org.apache.hadoop.hbase.CellUtil;
47 import org.apache.hadoop.hbase.HBaseConfiguration;
48 import org.apache.hadoop.hbase.HConstants;
49 import org.apache.hadoop.hbase.Stoppable;
50 import org.apache.hadoop.hbase.TableName;
51 import org.apache.hadoop.hbase.classification.InterfaceAudience;
52 import org.apache.hadoop.hbase.protobuf.generated.WALProtos.BulkLoadDescriptor;
53 import org.apache.hadoop.hbase.protobuf.generated.WALProtos.StoreDescriptor;
54 import org.apache.hadoop.hbase.regionserver.RSRpcServices;
55 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
56 import org.apache.hadoop.hbase.replication.ChainWALEntryFilter;
57 import org.apache.hadoop.hbase.replication.ClusterMarkingEntryFilter;
58 import org.apache.hadoop.hbase.replication.ReplicationEndpoint;
59 import org.apache.hadoop.hbase.replication.ReplicationException;
60 import org.apache.hadoop.hbase.replication.ReplicationPeer;
61 import org.apache.hadoop.hbase.replication.ReplicationPeers;
62 import org.apache.hadoop.hbase.replication.ReplicationQueueInfo;
63 import org.apache.hadoop.hbase.replication.ReplicationQueues;
64 import org.apache.hadoop.hbase.replication.SystemTableWALEntryFilter;
65 import org.apache.hadoop.hbase.replication.WALEntryFilter;
66 import org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceWALReaderThread.WALEntryBatch;
67 import org.apache.hadoop.hbase.util.Bytes;
68 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
69 import org.apache.hadoop.hbase.util.FSUtils;
70 import org.apache.hadoop.hbase.util.Pair;
71 import org.apache.hadoop.hbase.util.Threads;
72 import org.apache.hadoop.hbase.wal.DefaultWALProvider;
73 import org.apache.hadoop.hbase.wal.WAL.Entry;
74
75
76
77
78
79
80
81
82
83
84
85
86
87 @InterfaceAudience.Private
88 public class ReplicationSource extends Thread implements ReplicationSourceInterface {
89
90 private static final Log LOG = LogFactory.getLog(ReplicationSource.class);
91
92
93 private Map<String, PriorityBlockingQueue<Path>> queues =
94 new HashMap<String, PriorityBlockingQueue<Path>>();
95
96 private int queueSizePerGroup;
97 private ReplicationQueues replicationQueues;
98 private ReplicationPeers replicationPeers;
99
100 private Configuration conf;
101 private ReplicationQueueInfo replicationQueueInfo;
102
103 private String peerId;
104
105 private ReplicationSourceManager manager;
106
107 private Stoppable stopper;
108
109 private long sleepForRetries;
110 private FileSystem fs;
111
112 private UUID clusterId;
113
114 private UUID peerClusterId;
115
116 private AtomicLong totalReplicatedEdits = new AtomicLong(0);
117
118 private AtomicLong totalReplicatedOperations = new AtomicLong(0);
119
120 private String peerClusterZnode;
121
122 private int maxRetriesMultiplier;
123
124 private volatile boolean sourceRunning = false;
125
126 private MetricsSource metrics;
127
128 private int logQueueWarnThreshold;
129
130 private ReplicationEndpoint replicationEndpoint;
131
132 private WALEntryFilter walEntryFilter;
133
134 private ReplicationThrottler throttler;
135 private long defaultBandwidth;
136 private long currentBandwidth;
137 private ConcurrentHashMap<String, ReplicationSourceShipperThread> workerThreads =
138 new ConcurrentHashMap<String, ReplicationSourceShipperThread>();
139
140
141 public enum WorkerState {
142 RUNNING,
143 STOPPED,
144 FINISHED
145 }
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160 @Override
161 public void init(final Configuration conf, final FileSystem fs,
162 final ReplicationSourceManager manager, final ReplicationQueues replicationQueues,
163 final ReplicationPeers replicationPeers, final Stoppable stopper,
164 final String peerClusterZnode, final UUID clusterId, ReplicationEndpoint replicationEndpoint,
165 final MetricsSource metrics)
166 throws IOException {
167 this.stopper = stopper;
168 this.conf = HBaseConfiguration.create(conf);
169 decorateConf();
170 this.sleepForRetries =
171 this.conf.getLong("replication.source.sleepforretries", 1000);
172 this.maxRetriesMultiplier =
173 this.conf.getInt("replication.source.maxretriesmultiplier", 300);
174 this.queueSizePerGroup = this.conf.getInt("hbase.regionserver.maxlogs", 32);
175 this.replicationQueues = replicationQueues;
176 this.replicationPeers = replicationPeers;
177 this.manager = manager;
178 this.fs = fs;
179 this.metrics = metrics;
180 this.clusterId = clusterId;
181
182 this.peerClusterZnode = peerClusterZnode;
183 this.replicationQueueInfo = new ReplicationQueueInfo(peerClusterZnode);
184
185 this.peerId = this.replicationQueueInfo.getPeerId();
186 this.logQueueWarnThreshold = this.conf.getInt("replication.source.log.queue.warn", 2);
187 this.replicationEndpoint = replicationEndpoint;
188
189 defaultBandwidth = this.conf.getLong("replication.source.per.peer.node.bandwidth", 0);
190 currentBandwidth = getCurrentBandwidth();
191 this.throttler = new ReplicationThrottler((double) currentBandwidth / 10.0);
192
193 LOG.info("peerClusterZnode=" + peerClusterZnode + ", ReplicationSource : " + peerId
194 + ", currentBandwidth=" + this.currentBandwidth);
195 }
196
197 private void decorateConf() {
198 String replicationCodec = this.conf.get(HConstants.REPLICATION_CODEC_CONF_KEY);
199 if (StringUtils.isNotEmpty(replicationCodec)) {
200 this.conf.set(HConstants.RPC_CODEC_CONF_KEY, replicationCodec);
201 }
202 }
203
204 @Override
205 public void enqueueLog(Path log) {
206 String logPrefix = DefaultWALProvider.getWALPrefixFromWALName(log.getName());
207 PriorityBlockingQueue<Path> queue = queues.get(logPrefix);
208 if (queue == null) {
209 queue = new PriorityBlockingQueue<Path>(queueSizePerGroup, new LogsComparator());
210 queues.put(logPrefix, queue);
211 if (this.sourceRunning) {
212
213
214
215 final ReplicationSourceShipperThread worker =
216 new ReplicationSourceShipperThread(logPrefix, queue, replicationQueueInfo, this);
217 ReplicationSourceShipperThread extant = workerThreads.putIfAbsent(logPrefix, worker);
218 if (extant != null) {
219 LOG.debug("Someone has beat us to start a worker thread for wal group " + logPrefix);
220 } else {
221 LOG.debug("Starting up worker for wal group " + logPrefix);
222 worker.startup();
223 }
224 }
225 }
226 queue.put(log);
227 this.metrics.incrSizeOfLogQueue();
228
229 int queueSize = queue.size();
230 if (queueSize > this.logQueueWarnThreshold) {
231 LOG.warn("WAL group " + logPrefix + " queue size: " + queueSize
232 + " exceeds value of replication.source.log.queue.warn: " + logQueueWarnThreshold);
233 }
234 }
235
236 @Override
237 public void addHFileRefs(TableName tableName, byte[] family, List<Pair<Path, Path>> pairs)
238 throws ReplicationException {
239 String peerId = peerClusterZnode;
240 if (peerId.contains("-")) {
241
242
243 peerId = peerClusterZnode.split("-")[0];
244 }
245 Map<TableName, List<String>> tableCFMap = replicationPeers.getPeer(peerId).getTableCFs();
246 if (tableCFMap != null) {
247 List<String> tableCfs = tableCFMap.get(tableName);
248 if (tableCFMap.containsKey(tableName)
249 && (tableCfs == null || tableCfs.contains(Bytes.toString(family)))) {
250 this.replicationQueues.addHFileRefs(peerId, pairs);
251 metrics.incrSizeOfHFileRefsQueue(pairs.size());
252 } else {
253 LOG.debug("HFiles will not be replicated belonging to the table " + tableName + " family "
254 + Bytes.toString(family) + " to peer id " + peerId);
255 }
256 } else {
257
258
259 this.replicationQueues.addHFileRefs(peerId, pairs);
260 metrics.incrSizeOfHFileRefsQueue(pairs.size());
261 }
262 }
263
264 private void uninitialize() {
265 LOG.debug("Source exiting " + this.peerId);
266 metrics.clear();
267 if (replicationEndpoint.state() == Service.State.STARTING
268 || replicationEndpoint.state() == Service.State.RUNNING) {
269 replicationEndpoint.stopAndWait();
270 }
271 }
272
273 @Override
274 public void run() {
275
276 this.sourceRunning = true;
277 try {
278
279 Service.State state = replicationEndpoint.start().get();
280 if (state != Service.State.RUNNING) {
281 LOG.warn("ReplicationEndpoint was not started. Exiting");
282 uninitialize();
283 return;
284 }
285 } catch (Exception ex) {
286 LOG.warn("Error starting ReplicationEndpoint, exiting", ex);
287 throw new RuntimeException(ex);
288 }
289
290
291 ArrayList<WALEntryFilter> filters = Lists.newArrayList(
292 (WALEntryFilter)new SystemTableWALEntryFilter());
293 WALEntryFilter filterFromEndpoint = this.replicationEndpoint.getWALEntryfilter();
294 if (filterFromEndpoint != null) {
295 filters.add(filterFromEndpoint);
296 }
297 this.walEntryFilter = new ChainWALEntryFilter(filters);
298
299 int sleepMultiplier = 1;
300
301 while (this.isSourceActive() && this.peerClusterId == null) {
302 this.peerClusterId = replicationEndpoint.getPeerUUID();
303 if (this.isSourceActive() && this.peerClusterId == null) {
304 if (sleepForRetries("Cannot contact the peer's zk ensemble", sleepMultiplier)) {
305 sleepMultiplier++;
306 }
307 }
308 }
309
310
311
312 if (clusterId.equals(peerClusterId) && !replicationEndpoint.canReplicateToSameCluster()) {
313 this.terminate("ClusterId " + clusterId + " is replicating to itself: peerClusterId "
314 + peerClusterId + " which is not allowed by ReplicationEndpoint:"
315 + replicationEndpoint.getClass().getName(), null, false);
316 this.manager.closeQueue(this);
317 return;
318 }
319 LOG.info("Replicating " + clusterId + " -> " + peerClusterId);
320
321 for (Map.Entry<String, PriorityBlockingQueue<Path>> entry : queues.entrySet()) {
322 String walGroupId = entry.getKey();
323 PriorityBlockingQueue<Path> queue = entry.getValue();
324 final ReplicationSourceShipperThread worker =
325 new ReplicationSourceShipperThread(walGroupId, queue, replicationQueueInfo, this);
326 ReplicationSourceShipperThread extant = workerThreads.putIfAbsent(walGroupId, worker);
327 if (extant != null) {
328 LOG.debug("Someone has beat us to start a worker thread for wal group " + walGroupId);
329 } else {
330 LOG.debug("Starting up worker for wal group " + walGroupId);
331 worker.startup();
332 }
333 }
334 }
335
336
337
338
339
340
341
342 protected boolean sleepForRetries(String msg, int sleepMultiplier) {
343 try {
344 if (LOG.isTraceEnabled()) {
345 LOG.trace(msg + ", sleeping " + sleepForRetries + " times " + sleepMultiplier);
346 }
347 Thread.sleep(this.sleepForRetries * sleepMultiplier);
348 } catch (InterruptedException e) {
349 LOG.debug("Interrupted while sleeping between retries");
350 Thread.currentThread().interrupt();
351 }
352 return sleepMultiplier < maxRetriesMultiplier;
353 }
354
355
356
357
358
359
360 protected boolean isPeerEnabled() {
361 return this.replicationPeers.getStatusOfPeer(this.peerId);
362 }
363
364 @Override
365 public void startup() {
366 String n = Thread.currentThread().getName();
367 Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() {
368 @Override
369 public void uncaughtException(final Thread t, final Throwable e) {
370 LOG.error("Unexpected exception in ReplicationSource", e);
371 }
372 };
373 Threads
374 .setDaemonThreadRunning(this, n + ".replicationSource," + this.peerClusterZnode, handler);
375 }
376
377 @Override
378 public void terminate(String reason) {
379 terminate(reason, null);
380 }
381
382 @Override
383 public void terminate(String reason, Exception cause) {
384 terminate(reason, cause, true);
385 }
386
387 public void terminate(String reason, Exception cause, boolean join) {
388 if (cause == null) {
389 LOG.info("Closing source "
390 + this.peerClusterZnode + " because: " + reason);
391
392 } else {
393 LOG.error("Closing source " + this.peerClusterZnode
394 + " because an error occurred: " + reason, cause);
395 }
396 this.sourceRunning = false;
397 Collection<ReplicationSourceShipperThread> workers = workerThreads.values();
398 for (ReplicationSourceShipperThread worker : workers) {
399 worker.setWorkerState(WorkerState.STOPPED);
400 worker.entryReader.interrupt();
401 worker.interrupt();
402 }
403 ListenableFuture<Service.State> future = null;
404 if (this.replicationEndpoint != null) {
405 future = this.replicationEndpoint.stop();
406 }
407 if (join) {
408 for (ReplicationSourceShipperThread worker : workers) {
409 Threads.shutdown(worker, this.sleepForRetries);
410 LOG.info("ReplicationSourceWorker " + worker.getName() + " terminated");
411 }
412 if (future != null) {
413 try {
414 future.get(sleepForRetries * maxRetriesMultiplier, TimeUnit.MILLISECONDS);
415 } catch (Exception e) {
416 LOG.warn("Got exception while waiting for endpoint to shutdown for replication source :"
417 + this.peerClusterZnode,
418 e);
419 }
420 }
421 }
422 }
423
424 @Override
425 public String getPeerClusterZnode() {
426 return this.peerClusterZnode;
427 }
428
429 @Override
430 public String getPeerClusterId() {
431 return this.peerId;
432 }
433
434 @Override
435 public Path getCurrentPath() {
436
437 for (ReplicationSourceShipperThread worker : workerThreads.values()) {
438 if (worker.getCurrentPath() != null) return worker.getCurrentPath();
439 }
440 return null;
441 }
442
443 private boolean isSourceActive() {
444 return !this.stopper.isStopped() && this.sourceRunning;
445 }
446
447
448
449
450 public static class LogsComparator implements Comparator<Path> {
451
452 @Override
453 public int compare(Path o1, Path o2) {
454 return Long.compare(getTS(o1), getTS(o2));
455 }
456
457
458
459
460
461
462
463 private static long getTS(Path p) {
464 int tsIndex = p.getName().lastIndexOf('.') + 1;
465 return Long.parseLong(p.getName().substring(tsIndex));
466 }
467 }
468
469 @Override
470 public String getStats() {
471 StringBuilder sb = new StringBuilder();
472 sb.append("Total replicated edits: ").append(totalReplicatedEdits)
473 .append(", current progress: \n");
474 for (Map.Entry<String, ReplicationSourceShipperThread> entry : workerThreads.entrySet()) {
475 String walGroupId = entry.getKey();
476 ReplicationSourceShipperThread worker = entry.getValue();
477 long position = worker.getCurrentPosition();
478 Path currentPath = worker.getCurrentPath();
479 sb.append("walGroup [").append(walGroupId).append("]: ");
480 if (currentPath != null) {
481 sb.append("currently replicating from: ").append(currentPath).append(" at position: ")
482 .append(position).append("\n");
483 } else {
484 sb.append("no replication ongoing, waiting for new log");
485 }
486 }
487 return sb.toString();
488 }
489
490 @Override
491 public MetricsSource getSourceMetrics() {
492 return this.metrics;
493 }
494
495 private long getCurrentBandwidth() {
496 ReplicationPeer replicationPeer = this.replicationPeers.getPeer(peerId);
497 long peerBandwidth = replicationPeer != null ? replicationPeer.getPeerBandwidth() : 0;
498
499 return peerBandwidth != 0 ? peerBandwidth : defaultBandwidth;
500 }
501
502
503
504 public class ReplicationSourceShipperThread extends Thread {
505 ReplicationSourceInterface source;
506 String walGroupId;
507 PriorityBlockingQueue<Path> queue;
508 ReplicationQueueInfo replicationQueueInfo;
509
510 private long lastLoggedPosition = -1;
511
512 private volatile Path currentPath;
513
514 private WorkerState state;
515 ReplicationSourceWALReaderThread entryReader;
516
517 public ReplicationSourceShipperThread(String walGroupId,
518 PriorityBlockingQueue<Path> queue, ReplicationQueueInfo replicationQueueInfo,
519 ReplicationSourceInterface source) {
520 this.walGroupId = walGroupId;
521 this.queue = queue;
522 this.replicationQueueInfo = replicationQueueInfo;
523 this.source = source;
524 }
525
526 @Override
527 public void run() {
528 setWorkerState(WorkerState.RUNNING);
529
530 while (isWorkerActive()) {
531 int sleepMultiplier = 1;
532
533 if (!isPeerEnabled()) {
534 if (sleepForRetries("Replication is disabled", sleepMultiplier)) {
535 sleepMultiplier++;
536 }
537 continue;
538 }
539 while (entryReader == null) {
540 if (sleepForRetries("Replication WAL entry reader thread not initialized",
541 sleepMultiplier)) {
542 sleepMultiplier++;
543 }
544 if (sleepMultiplier == maxRetriesMultiplier) {
545 LOG.warn("Replication WAL entry reader thread not initialized");
546 }
547 }
548
549 try {
550 WALEntryBatch entryBatch = entryReader.take();
551 shipEdits(entryBatch);
552 if (replicationQueueInfo.isQueueRecovered() && entryBatch.getWalEntries().isEmpty()
553 && entryBatch.getLastSeqIds().isEmpty()) {
554 LOG.debug("Finished recovering queue for group " + walGroupId + " of peer "
555 + peerClusterZnode);
556 metrics.incrCompletedRecoveryQueue();
557 setWorkerState(WorkerState.FINISHED);
558 continue;
559 }
560 } catch (InterruptedException e) {
561 LOG.trace("Interrupted while waiting for next replication entry batch", e);
562 Thread.currentThread().interrupt();
563 }
564 }
565
566 if (replicationQueueInfo.isQueueRecovered() && getWorkerState() == WorkerState.FINISHED) {
567
568 synchronized (this) {
569 Threads.sleep(100);
570 boolean allOtherTaskDone = true;
571 for (ReplicationSourceShipperThread worker : workerThreads.values()) {
572 if (!worker.equals(this) && worker.getWorkerState() != WorkerState.FINISHED) {
573 allOtherTaskDone = false;
574 break;
575 }
576 }
577 if (allOtherTaskDone) {
578 manager.closeRecoveredQueue(this.source);
579
580 if (source instanceof ReplicationSource) {
581 ((ReplicationSource) source).replicationEndpoint.stop();
582 }
583 LOG.info("Finished recovering queue " + peerClusterZnode
584 + " with the following stats: " + getStats());
585 }
586 }
587 }
588
589 if (state != WorkerState.FINISHED) {
590 setWorkerState(WorkerState.STOPPED);
591 }
592 }
593
594 private void cleanUpHFileRefs(WALEdit edit) throws IOException {
595 String peerId = peerClusterZnode;
596 if (peerId.contains("-")) {
597
598
599 peerId = peerClusterZnode.split("-")[0];
600 }
601 List<Cell> cells = edit.getCells();
602 int totalCells = cells.size();
603 for (int i = 0; i < totalCells; i++) {
604 Cell cell = cells.get(i);
605 if (CellUtil.matchingQualifier(cell, WALEdit.BULK_LOAD)) {
606 BulkLoadDescriptor bld = WALEdit.getBulkLoadDescriptor(cell);
607 List<StoreDescriptor> stores = bld.getStoresList();
608 int totalStores = stores.size();
609 for (int j = 0; j < totalStores; j++) {
610 List<String> storeFileList = stores.get(j).getStoreFileList();
611 manager.cleanUpHFileRefs(peerId, storeFileList);
612 metrics.decrSizeOfHFileRefsQueue(storeFileList.size());
613 }
614 }
615 }
616 }
617
618 private void checkBandwidthChangeAndResetThrottler() {
619 long peerBandwidth = getCurrentBandwidth();
620 if (peerBandwidth != currentBandwidth) {
621 currentBandwidth = peerBandwidth;
622 throttler.setBandwidth((double) currentBandwidth / 10.0);
623 LOG.info("ReplicationSource : " + peerId
624 + " bandwidth throttling changed, currentBandWidth=" + currentBandwidth);
625 }
626 }
627
628
629
630
631 protected void shipEdits(WALEntryBatch entryBatch) {
632 List<Entry> entries = entryBatch.getWalEntries();
633 long lastReadPosition = entryBatch.getLastWalPosition();
634 currentPath = entryBatch.getLastWalPath();
635 int sleepMultiplier = 0;
636 if (entries.isEmpty()) {
637 if (lastLoggedPosition != lastReadPosition) {
638 updateLogPosition(lastReadPosition);
639
640
641 metrics.setAgeOfLastShippedOp(EnvironmentEdgeManager.currentTime(), walGroupId);
642 }
643 return;
644 }
645 int currentSize = (int) entryBatch.getHeapSize();
646 while (isWorkerActive()) {
647 try {
648 checkBandwidthChangeAndResetThrottler();
649 if (throttler.isEnabled()) {
650 long sleepTicks = throttler.getNextSleepInterval(currentSize);
651 if (sleepTicks > 0) {
652 try {
653 if (LOG.isTraceEnabled()) {
654 LOG.trace("To sleep " + sleepTicks + "ms for throttling control");
655 }
656 Thread.sleep(sleepTicks);
657 } catch (InterruptedException e) {
658 LOG.debug("Interrupted while sleeping for throttling control");
659 Thread.currentThread().interrupt();
660
661
662 continue;
663 }
664
665 throttler.resetStartTick();
666 }
667 }
668
669
670 ReplicationEndpoint.ReplicateContext replicateContext =
671 new ReplicationEndpoint.ReplicateContext();
672 replicateContext.setEntries(entries).setSize(currentSize);
673 replicateContext.setWalGroupId(walGroupId);
674
675 long startTimeNs = System.nanoTime();
676
677 boolean replicated = replicationEndpoint.replicate(replicateContext);
678 long endTimeNs = System.nanoTime();
679
680 if (!replicated) {
681 continue;
682 } else {
683 sleepMultiplier = Math.max(sleepMultiplier - 1, 0);
684 }
685
686 if (this.lastLoggedPosition != lastReadPosition) {
687
688 int size = entries.size();
689 for (int i = 0; i < size; i++) {
690 cleanUpHFileRefs(entries.get(i).getEdit());
691 }
692
693 updateLogPosition(lastReadPosition);
694 }
695 if (throttler.isEnabled()) {
696 throttler.addPushSize(currentSize);
697 }
698 totalReplicatedEdits.addAndGet(entries.size());
699 totalReplicatedOperations.addAndGet(entryBatch.getNbOperations());
700
701 metrics.shipBatch(entryBatch.getNbOperations(), currentSize, entryBatch.getNbHFiles());
702 metrics.setAgeOfLastShippedOp(entries.get(entries.size() - 1).getKey().getWriteTime(),
703 walGroupId);
704 if (LOG.isTraceEnabled()) {
705 LOG.trace("Replicated " + totalReplicatedEdits + " entries in total, or "
706 + totalReplicatedOperations + " operations in "
707 + ((endTimeNs - startTimeNs) / 1000000) + " ms");
708 }
709 break;
710 } catch (Exception ex) {
711 LOG.warn(replicationEndpoint.getClass().getName() + " threw unknown exception:"
712 + org.apache.hadoop.util.StringUtils.stringifyException(ex));
713 if (sleepForRetries("ReplicationEndpoint threw exception", sleepMultiplier)) {
714 sleepMultiplier++;
715 }
716 }
717 }
718 }
719
720 private void updateLogPosition(long lastReadPosition) {
721 manager.logPositionAndCleanOldLogs(currentPath, peerClusterZnode, lastReadPosition,
722 this.replicationQueueInfo.isQueueRecovered(), false);
723 lastLoggedPosition = lastReadPosition;
724 }
725
726 public void startup() {
727 String n = Thread.currentThread().getName();
728 Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() {
729 @Override
730 public void uncaughtException(final Thread t, final Throwable e) {
731 RSRpcServices.exitIfOOME(e);
732 LOG.error("Unexpected exception in ReplicationSourceWorkerThread," + " currentPath="
733 + getCurrentPath(), e);
734 stopper.stop("Unexpected exception in ReplicationSourceWorkerThread");
735 }
736 };
737 Threads.setDaemonThreadRunning(this, n + ".replicationSource." + walGroupId + ","
738 + peerClusterZnode, handler);
739 workerThreads.put(walGroupId, this);
740
741 long startPosition = 0;
742
743 if (this.replicationQueueInfo.isQueueRecovered()) {
744 startPosition = getRecoveredQueueStartPos(startPosition);
745 int numRetries = 0;
746 while (numRetries <= maxRetriesMultiplier) {
747 try {
748 locateRecoveredPaths();
749 break;
750 } catch (IOException e) {
751 LOG.error("Error while locating recovered queue paths, attempt #" + numRetries);
752 numRetries++;
753 }
754 }
755 }
756
757 startWALReaderThread(n, handler, startPosition);
758 }
759
760
761
762 private long getRecoveredQueueStartPos(long startPosition) {
763 try {
764 startPosition =
765 (replicationQueues.getLogPosition(peerClusterZnode, this.queue.peek().getName()));
766 if (LOG.isTraceEnabled()) {
767 LOG.trace("Recovered queue started with log " + this.queue.peek() + " at position "
768 + startPosition);
769 }
770 } catch (ReplicationException e) {
771 terminate("Couldn't get the position of this recovered queue " + peerClusterZnode, e);
772 }
773 return startPosition;
774 }
775
776
777 private void startWALReaderThread(String threadName, Thread.UncaughtExceptionHandler handler,
778 long startPosition) {
779 ArrayList<WALEntryFilter> filters = Lists.newArrayList(walEntryFilter,
780 new ClusterMarkingEntryFilter(clusterId, peerClusterId, replicationEndpoint));
781 ChainWALEntryFilter readerFilter = new ChainWALEntryFilter(filters);
782 entryReader = new ReplicationSourceWALReaderThread(manager, replicationQueueInfo, queue,
783 startPosition, fs, conf, readerFilter, metrics);
784 Threads.setDaemonThreadRunning(entryReader, threadName
785 + ".replicationSource.replicationWALReaderThread." + walGroupId + "," + peerClusterZnode,
786 handler);
787 }
788
789
790
791 private void locateRecoveredPaths() throws IOException {
792 boolean hasPathChanged = false;
793 PriorityBlockingQueue<Path> newPaths =
794 new PriorityBlockingQueue<Path>(queueSizePerGroup, new LogsComparator());
795 pathsLoop: for (Path path : queue) {
796 if (fs.exists(path)) {
797 newPaths.add(path);
798 continue;
799 }
800
801 hasPathChanged = true;
802 if (stopper instanceof ReplicationSyncUp.DummyServer) {
803
804
805 Path newPath = getReplSyncUpPath(path);
806 newPaths.add(newPath);
807 continue;
808 } else {
809
810
811 List<String> deadRegionServers = this.replicationQueueInfo.getDeadRegionServers();
812 LOG.info("NB dead servers : " + deadRegionServers.size());
813 final Path walDir = FSUtils.getWALRootDir(conf);
814 for (String curDeadServerName : deadRegionServers) {
815 final Path deadRsDirectory =
816 new Path(walDir, DefaultWALProvider.getWALDirectoryName(curDeadServerName));
817 Path[] locs = new Path[] { new Path(deadRsDirectory, path.getName()), new Path(
818 deadRsDirectory.suffix(DefaultWALProvider.SPLITTING_EXT), path.getName()) };
819 for (Path possibleLogLocation : locs) {
820 LOG.info("Possible location " + possibleLogLocation.toUri().toString());
821 if (manager.getFs().exists(possibleLogLocation)) {
822
823 LOG.info("Log " + path + " still exists at " + possibleLogLocation);
824 newPaths.add(possibleLogLocation);
825 continue pathsLoop;
826 }
827 }
828 }
829
830 LOG.error(
831 String.format("WAL Path %s doesn't exist and couldn't find its new location", path));
832 newPaths.add(path);
833 }
834 }
835
836 if (hasPathChanged) {
837 if (newPaths.size() != queue.size()) {
838 LOG.error("Recovery queue size is incorrect");
839 throw new IOException("Recovery queue size error");
840 }
841
842
843
844 queue.clear();
845 for (Path path : newPaths) {
846 queue.add(path);
847 }
848 }
849 }
850
851
852
853 private Path getReplSyncUpPath(Path path) throws IOException {
854 FileStatus[] rss = fs.listStatus(manager.getLogDir());
855 for (FileStatus rs : rss) {
856 Path p = rs.getPath();
857 FileStatus[] logs = fs.listStatus(p);
858 for (FileStatus log : logs) {
859 p = new Path(p, log.getPath().getName());
860 if (p.getName().equals(path.getName())) {
861 LOG.info("Log " + p.getName() + " found at " + p);
862 return p;
863 }
864 }
865 }
866 LOG.error("Didn't find path for: " + path.getName());
867 return path;
868 }
869
870 public Path getCurrentPath() {
871 return this.entryReader.getCurrentPath();
872 }
873
874 public long getCurrentPosition() {
875 return this.lastLoggedPosition;
876 }
877
878 private boolean isWorkerActive() {
879 return !stopper.isStopped() && state == WorkerState.RUNNING && !isInterrupted();
880 }
881
882 private void terminate(String reason, Exception cause) {
883 if (cause == null) {
884 LOG.info("Closing worker for wal group " + this.walGroupId + " because: " + reason);
885
886 } else {
887 LOG.error("Closing worker for wal group " + this.walGroupId
888 + " because an error occurred: " + reason, cause);
889 }
890 entryReader.interrupt();
891 Threads.shutdown(entryReader, sleepForRetries);
892 setWorkerState(WorkerState.STOPPED);
893 this.interrupt();
894 Threads.shutdown(this, sleepForRetries);
895 LOG.info("ReplicationSourceWorker " + this.getName() + " terminated");
896 }
897
898
899
900
901
902 public void setWorkerState(WorkerState state) {
903 this.state = state;
904 if (entryReader != null) {
905 entryReader.setReaderRunning(state == WorkerState.RUNNING);
906 }
907 }
908
909
910
911
912
913 public WorkerState getWorkerState() {
914 return state;
915 }
916 }
917 }