1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.regionserver;
20
21 import static org.apache.hadoop.util.StringUtils.humanReadableInt;
22
23 import java.io.IOException;
24 import java.lang.Thread.UncaughtExceptionHandler;
25 import java.lang.management.MemoryUsage;
26 import java.util.ArrayList;
27 import java.util.ConcurrentModificationException;
28 import java.util.HashMap;
29 import java.util.HashSet;
30 import java.util.List;
31 import java.util.Map;
32 import java.util.Set;
33 import java.util.SortedMap;
34 import java.util.concurrent.BlockingQueue;
35 import java.util.concurrent.DelayQueue;
36 import java.util.concurrent.Delayed;
37 import java.util.concurrent.ThreadFactory;
38 import java.util.concurrent.TimeUnit;
39 import java.util.concurrent.atomic.AtomicBoolean;
40 import java.util.concurrent.locks.ReentrantReadWriteLock;
41
42 import org.apache.commons.logging.Log;
43 import org.apache.commons.logging.LogFactory;
44 import org.apache.hadoop.conf.Configuration;
45 import org.apache.hadoop.hbase.DroppedSnapshotException;
46 import org.apache.hadoop.hbase.HConstants;
47 import org.apache.hadoop.hbase.RemoteExceptionHandler;
48 import org.apache.hadoop.hbase.classification.InterfaceAudience;
49 import org.apache.hadoop.hbase.client.RegionReplicaUtil;
50 import org.apache.hadoop.hbase.io.util.HeapMemorySizeUtil;
51 import org.apache.hadoop.hbase.regionserver.Region.FlushResult;
52 import org.apache.hadoop.hbase.util.Bytes;
53 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
54 import org.apache.hadoop.hbase.util.HasThread;
55 import org.apache.hadoop.hbase.util.ServerRegionReplicaUtil;
56 import org.apache.hadoop.hbase.util.Threads;
57 import org.apache.hadoop.util.StringUtils;
58 import org.apache.hadoop.util.StringUtils.TraditionalBinaryPrefix;
59 import org.apache.htrace.Trace;
60 import org.apache.htrace.TraceScope;
61 import org.apache.hadoop.hbase.util.Counter;
62
63 import com.google.common.base.Preconditions;
64
65
66
67
68
69
70
71
72
73
74 @InterfaceAudience.Private
75 class MemStoreFlusher implements FlushRequester {
76 private static final Log LOG = LogFactory.getLog(MemStoreFlusher.class);
77
78 private Configuration conf;
79
80
81 private final BlockingQueue<FlushQueueEntry> flushQueue =
82 new DelayQueue<FlushQueueEntry>();
83 private final Map<Region, FlushRegionEntry> regionsInQueue =
84 new HashMap<Region, FlushRegionEntry>();
85 private AtomicBoolean wakeupPending = new AtomicBoolean();
86
87 private final long threadWakeFrequency;
88 private final HRegionServer server;
89 private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
90 private final Object blockSignal = new Object();
91
92 protected long globalMemStoreLimit;
93 protected float globalMemStoreLimitLowMarkPercent;
94 protected long globalMemStoreLimitLowMark;
95
96 private long blockingWaitTime;
97 private final Counter updatesBlockedMsHighWater = new Counter();
98
99 private final FlushHandler[] flushHandlers;
100 private List<FlushRequestListener> flushRequestListeners = new ArrayList<FlushRequestListener>(1);
101
102
103
104
105
106 public MemStoreFlusher(final Configuration conf,
107 final HRegionServer server) {
108 super();
109 this.conf = conf;
110 this.server = server;
111 this.threadWakeFrequency =
112 conf.getLong(HConstants.THREAD_WAKE_FREQUENCY, 10 * 1000);
113 long max = -1L;
114 final MemoryUsage usage = HeapMemorySizeUtil.safeGetHeapMemoryUsage();
115 if (usage != null) {
116 max = usage.getMax();
117 }
118 float globalMemStorePercent = HeapMemorySizeUtil.getGlobalMemStorePercent(conf, true);
119 this.globalMemStoreLimit = (long) (max * globalMemStorePercent);
120 this.globalMemStoreLimitLowMarkPercent =
121 HeapMemorySizeUtil.getGlobalMemStoreLowerMark(conf, globalMemStorePercent);
122 this.globalMemStoreLimitLowMark =
123 (long) (this.globalMemStoreLimit * this.globalMemStoreLimitLowMarkPercent);
124
125 this.blockingWaitTime = conf.getInt("hbase.hstore.blockingWaitTime",
126 90000);
127 int handlerCount = conf.getInt("hbase.hstore.flusher.count", 2);
128 this.flushHandlers = new FlushHandler[handlerCount];
129 LOG.info("globalMemStoreLimit="
130 + TraditionalBinaryPrefix.long2String(this.globalMemStoreLimit, "", 1)
131 + ", globalMemStoreLimitLowMark="
132 + TraditionalBinaryPrefix.long2String(this.globalMemStoreLimitLowMark, "", 1)
133 + ", maxHeap=" + TraditionalBinaryPrefix.long2String(max, "", 1));
134 }
135
136 public Counter getUpdatesBlockedMsHighWater() {
137 return this.updatesBlockedMsHighWater;
138 }
139
140
141
142
143
144
145
146 private boolean flushOneForGlobalPressure() {
147 SortedMap<Long, Region> regionsBySize = server.getCopyOfOnlineRegionsSortedBySize();
148 Set<Region> excludedRegions = new HashSet<Region>();
149
150 double secondaryMultiplier
151 = ServerRegionReplicaUtil.getRegionReplicaStoreFileRefreshMultiplier(conf);
152
153 boolean flushedOne = false;
154 while (!flushedOne) {
155
156
157 Region bestFlushableRegion = getBiggestMemstoreRegion(regionsBySize, excludedRegions, true);
158
159 Region bestAnyRegion = getBiggestMemstoreRegion(
160 regionsBySize, excludedRegions, false);
161
162 Region bestRegionReplica = getBiggestMemstoreOfRegionReplica(regionsBySize,
163 excludedRegions);
164
165 if (bestAnyRegion == null && bestRegionReplica == null) {
166 LOG.error("Above memory mark but there are no flushable regions!");
167 return false;
168 }
169
170 Region regionToFlush;
171 if (bestFlushableRegion != null &&
172 bestAnyRegion.getMemstoreSize() > 2 * bestFlushableRegion.getMemstoreSize()) {
173
174
175
176
177 if (LOG.isDebugEnabled()) {
178 LOG.debug("Under global heap pressure: " + "Region "
179 + bestAnyRegion.getRegionInfo().getRegionNameAsString()
180 + " has too many " + "store files, but is "
181 + TraditionalBinaryPrefix.long2String(bestAnyRegion.getMemstoreSize(), "", 1)
182 + " vs best flushable region's "
183 + TraditionalBinaryPrefix.long2String(bestFlushableRegion.getMemstoreSize(), "", 1)
184 + ". Choosing the bigger.");
185 }
186 regionToFlush = bestAnyRegion;
187 } else {
188 if (bestFlushableRegion == null) {
189 regionToFlush = bestAnyRegion;
190 } else {
191 regionToFlush = bestFlushableRegion;
192 }
193 }
194
195 Preconditions.checkState(
196 (regionToFlush != null && regionToFlush.getMemstoreSize() > 0) ||
197 (bestRegionReplica != null && bestRegionReplica.getMemstoreSize() > 0));
198
199 if (regionToFlush == null ||
200 (bestRegionReplica != null &&
201 ServerRegionReplicaUtil.isRegionReplicaStoreFileRefreshEnabled(conf) &&
202 (bestRegionReplica.getMemstoreSize()
203 > secondaryMultiplier * regionToFlush.getMemstoreSize()))) {
204 LOG.info("Refreshing storefiles of region " + bestRegionReplica +
205 " due to global heap pressure. memstore size=" + StringUtils.humanReadableInt(
206 server.getRegionServerAccounting().getGlobalMemstoreSize()));
207 flushedOne = refreshStoreFilesAndReclaimMemory(bestRegionReplica);
208 if (!flushedOne) {
209 LOG.info("Excluding secondary region " + bestRegionReplica +
210 " - trying to find a different region to refresh files.");
211 excludedRegions.add(bestRegionReplica);
212 }
213 } else {
214 LOG.info("Flush of region " + regionToFlush + " due to global heap pressure. "
215 + "Total Memstore size="
216 + humanReadableInt(server.getRegionServerAccounting().getGlobalMemstoreSize())
217 + ", Region memstore size="
218 + humanReadableInt(regionToFlush.getMemstoreSize()));
219 flushedOne = flushRegion(regionToFlush, true, false);
220
221 if (!flushedOne) {
222 LOG.info("Excluding unflushable region " + regionToFlush +
223 " - trying to find a different region to flush.");
224 excludedRegions.add(regionToFlush);
225 }
226 }
227 }
228 return true;
229 }
230
231 private class FlushHandler extends HasThread {
232
233 private FlushHandler(String name) {
234 super(name);
235 }
236
237 @Override
238 public void run() {
239 while (!server.isStopped()) {
240 FlushQueueEntry fqe = null;
241 try {
242 wakeupPending.set(false);
243 fqe = flushQueue.poll(threadWakeFrequency, TimeUnit.MILLISECONDS);
244 if (fqe == null || fqe instanceof WakeupFlushThread) {
245 if (isAboveLowWaterMark()) {
246 LOG.debug("Flush thread woke up because memory above low water="
247 + TraditionalBinaryPrefix.long2String(globalMemStoreLimitLowMark, "", 1));
248 if (!flushOneForGlobalPressure()) {
249
250
251
252
253
254 Thread.sleep(1000);
255 wakeUpIfBlocking();
256 }
257
258 wakeupFlushThread();
259 }
260 continue;
261 }
262 FlushRegionEntry fre = (FlushRegionEntry) fqe;
263 if (!flushRegion(fre)) {
264 break;
265 }
266 } catch (InterruptedException ex) {
267 continue;
268 } catch (ConcurrentModificationException ex) {
269 continue;
270 } catch (Exception ex) {
271 LOG.error("Cache flusher failed for entry " + fqe, ex);
272 if (!server.checkFileSystem()) {
273 break;
274 }
275 }
276 }
277 synchronized (regionsInQueue) {
278 regionsInQueue.clear();
279 flushQueue.clear();
280 }
281
282
283 wakeUpIfBlocking();
284 LOG.info(getName() + " exiting");
285 }
286 }
287
288
289 private void wakeupFlushThread() {
290 if (wakeupPending.compareAndSet(false, true)) {
291 flushQueue.add(new WakeupFlushThread());
292 }
293 }
294
295 private Region getBiggestMemstoreRegion(
296 SortedMap<Long, Region> regionsBySize,
297 Set<Region> excludedRegions,
298 boolean checkStoreFileCount) {
299 synchronized (regionsInQueue) {
300 for (Region region : regionsBySize.values()) {
301 if (excludedRegions.contains(region)) {
302 continue;
303 }
304
305 if (((HRegion)region).writestate.flushing ||
306 !((HRegion)region).writestate.writesEnabled) {
307 continue;
308 }
309
310 if (checkStoreFileCount && isTooManyStoreFiles(region)) {
311 continue;
312 }
313 return region;
314 }
315 }
316 return null;
317 }
318
319 private Region getBiggestMemstoreOfRegionReplica(SortedMap<Long, Region> regionsBySize,
320 Set<Region> excludedRegions) {
321 synchronized (regionsInQueue) {
322 for (Region region : regionsBySize.values()) {
323 if (excludedRegions.contains(region)) {
324 continue;
325 }
326
327 if (RegionReplicaUtil.isDefaultReplica(region.getRegionInfo())) {
328 continue;
329 }
330
331 return region;
332 }
333 }
334 return null;
335 }
336
337 private boolean refreshStoreFilesAndReclaimMemory(Region region) {
338 try {
339 return region.refreshStoreFiles();
340 } catch (IOException e) {
341 LOG.warn("Refreshing store files failed with exception", e);
342 }
343 return false;
344 }
345
346
347
348
349 private boolean isAboveHighWaterMark() {
350 return server.getRegionServerAccounting().
351 getGlobalMemstoreSize() >= globalMemStoreLimit;
352 }
353
354
355
356
357 private boolean isAboveLowWaterMark() {
358 return server.getRegionServerAccounting().
359 getGlobalMemstoreSize() >= globalMemStoreLimitLowMark;
360 }
361
362 @Override
363 public boolean requestFlush(Region r, boolean forceFlushAllStores) {
364 synchronized (regionsInQueue) {
365 if (!regionsInQueue.containsKey(r)) {
366
367
368 FlushRegionEntry fqe = new FlushRegionEntry(r, forceFlushAllStores);
369 this.regionsInQueue.put(r, fqe);
370 this.flushQueue.add(fqe);
371 ((HRegion)r).incrementFlushesQueuedCount();
372 return true;
373 }
374 return false;
375 }
376 }
377
378 @Override
379 public boolean requestDelayedFlush(Region r, long delay, boolean forceFlushAllStores) {
380 synchronized (regionsInQueue) {
381 if (!regionsInQueue.containsKey(r)) {
382
383 FlushRegionEntry fqe = new FlushRegionEntry(r, forceFlushAllStores);
384 fqe.requeue(delay);
385 this.regionsInQueue.put(r, fqe);
386 this.flushQueue.add(fqe);
387 ((HRegion)r).incrementFlushesQueuedCount();
388 return true;
389 }
390 return false;
391 }
392 }
393
394 public int getFlushQueueSize() {
395 return flushQueue.size();
396 }
397
398
399
400
401 void interruptIfNecessary() {
402 lock.writeLock().lock();
403 try {
404 for (FlushHandler flushHander : flushHandlers) {
405 if (flushHander != null) flushHander.interrupt();
406 }
407 } finally {
408 lock.writeLock().unlock();
409 }
410 }
411
412 synchronized void start(UncaughtExceptionHandler eh) {
413 ThreadFactory flusherThreadFactory = Threads.newDaemonThreadFactory(
414 server.getServerName().toShortString() + "-MemStoreFlusher", eh);
415 for (int i = 0; i < flushHandlers.length; i++) {
416 flushHandlers[i] = new FlushHandler("MemStoreFlusher." + i);
417 flusherThreadFactory.newThread(flushHandlers[i]);
418 flushHandlers[i].start();
419 }
420 }
421
422 boolean isAlive() {
423 for (FlushHandler flushHander : flushHandlers) {
424 if (flushHander != null && flushHander.isAlive()) {
425 return true;
426 }
427 }
428 return false;
429 }
430
431 void join() {
432 for (FlushHandler flushHander : flushHandlers) {
433 if (flushHander != null) {
434 Threads.shutdown(flushHander.getThread());
435 }
436 }
437 }
438
439
440
441
442
443
444
445
446
447 private boolean flushRegion(final FlushRegionEntry fqe) {
448 Region region = fqe.region;
449 if (!region.getRegionInfo().isMetaRegion() &&
450 isTooManyStoreFiles(region)) {
451 if (fqe.isMaximumWait(this.blockingWaitTime)) {
452 LOG.info("Waited " + (EnvironmentEdgeManager.currentTime() - fqe.createTime) +
453 "ms on a compaction to clean up 'too many store files'; waited " +
454 "long enough... proceeding with flush of " +
455 region.getRegionInfo().getRegionNameAsString());
456 } else {
457
458 if (fqe.getRequeueCount() <= 0) {
459
460 LOG.warn("Region " + region.getRegionInfo().getRegionNameAsString() + " has too many " +
461 "store files; delaying flush up to " + this.blockingWaitTime + "ms");
462 if (!this.server.compactSplitThread.requestSplit(region)) {
463 try {
464 this.server.compactSplitThread.requestSystemCompaction(
465 region, Thread.currentThread().getName());
466 } catch (IOException e) {
467 LOG.error("Cache flush failed for region " +
468 Bytes.toStringBinary(region.getRegionInfo().getRegionName()),
469 RemoteExceptionHandler.checkIOException(e));
470 }
471 }
472 }
473
474
475
476 this.flushQueue.add(fqe.requeue(this.blockingWaitTime / 100));
477
478 return true;
479 }
480 }
481 return flushRegion(region, false, fqe.isForceFlushAllStores());
482 }
483
484
485
486
487
488
489
490
491
492
493
494
495
496 private boolean flushRegion(final Region region, final boolean emergencyFlush,
497 boolean forceFlushAllStores) {
498 synchronized (this.regionsInQueue) {
499 FlushRegionEntry fqe = this.regionsInQueue.remove(region);
500
501 if (fqe != null && emergencyFlush) {
502
503
504 flushQueue.remove(fqe);
505 }
506 }
507
508 lock.readLock().lock();
509 try {
510 notifyFlushRequest(region, emergencyFlush);
511 FlushResult flushResult = region.flush(forceFlushAllStores);
512 boolean shouldCompact = flushResult.isCompactionNeeded();
513
514 boolean shouldSplit = ((HRegion)region).checkSplit() != null;
515 if (shouldSplit) {
516 this.server.compactSplitThread.requestSplit(region);
517 } else if (shouldCompact) {
518 server.compactSplitThread.requestSystemCompaction(
519 region, Thread.currentThread().getName());
520 }
521 } catch (DroppedSnapshotException ex) {
522
523
524
525
526
527 server.abort("Replay of WAL required. Forcing server shutdown", ex);
528 return false;
529 } catch (IOException ex) {
530 LOG.error("Cache flush failed" + (region != null ? (" for region " +
531 Bytes.toStringBinary(region.getRegionInfo().getRegionName())) : ""),
532 RemoteExceptionHandler.checkIOException(ex));
533 if (!server.checkFileSystem()) {
534 return false;
535 }
536 } finally {
537 lock.readLock().unlock();
538 wakeUpIfBlocking();
539 }
540 return true;
541 }
542
543 private void notifyFlushRequest(Region region, boolean emergencyFlush) {
544 FlushType type = FlushType.NORMAL;
545 if (emergencyFlush) {
546 type = isAboveHighWaterMark() ? FlushType.ABOVE_HIGHER_MARK : FlushType.ABOVE_LOWER_MARK;
547 }
548 for (FlushRequestListener listener : flushRequestListeners) {
549 listener.flushRequested(type, region);
550 }
551 }
552
553 private void wakeUpIfBlocking() {
554 synchronized (blockSignal) {
555 blockSignal.notifyAll();
556 }
557 }
558
559 private boolean isTooManyStoreFiles(Region region) {
560
561
562 if (!region.getTableDesc().isCompactionEnabled()) {
563 return false;
564 }
565
566 for (Store store : region.getStores()) {
567 if (store.hasTooManyStoreFiles()) {
568 return true;
569 }
570 }
571 return false;
572 }
573
574
575
576
577
578
579
580 public void reclaimMemStoreMemory() {
581 TraceScope scope = Trace.startSpan("MemStoreFluser.reclaimMemStoreMemory");
582 if (isAboveHighWaterMark()) {
583 if (Trace.isTracing()) {
584 scope.getSpan().addTimelineAnnotation("Force Flush. We're above high water mark.");
585 }
586 long start = EnvironmentEdgeManager.currentTime();
587 synchronized (this.blockSignal) {
588 boolean blocked = false;
589 long startTime = 0;
590 boolean interrupted = false;
591 try {
592 while (isAboveHighWaterMark() && !server.isStopped()) {
593 if (!blocked) {
594 startTime = EnvironmentEdgeManager.currentTime();
595 LOG.info("Blocking updates on "
596 + server.toString()
597 + ": the global memstore size "
598 + TraditionalBinaryPrefix.long2String(server.getRegionServerAccounting()
599 .getGlobalMemstoreSize(), "", 1) + " is >= than blocking "
600 + TraditionalBinaryPrefix.long2String(globalMemStoreLimit, "", 1) + " size");
601 }
602 blocked = true;
603 wakeupFlushThread();
604 try {
605
606
607 blockSignal.wait(5 * 1000);
608 } catch (InterruptedException ie) {
609 LOG.warn("Interrupted while waiting");
610 interrupted = true;
611 }
612 long took = EnvironmentEdgeManager.currentTime() - start;
613 LOG.warn("Memstore is above high water mark and block " + took + "ms");
614 }
615 } finally {
616 if (interrupted) {
617 Thread.currentThread().interrupt();
618 }
619 }
620
621 if(blocked){
622 final long totalTime = EnvironmentEdgeManager.currentTime() - startTime;
623 if(totalTime > 0){
624 this.updatesBlockedMsHighWater.add(totalTime);
625 }
626 LOG.info("Unblocking updates for server " + server.toString());
627 }
628 }
629 } else if (isAboveLowWaterMark()) {
630 wakeupFlushThread();
631 }
632 scope.close();
633 }
634 @Override
635 public String toString() {
636 return "flush_queue="
637 + flushQueue.size();
638 }
639
640 public String dumpQueue() {
641 StringBuilder queueList = new StringBuilder();
642 queueList.append("Flush Queue Queue dump:\n");
643 queueList.append(" Flush Queue:\n");
644 java.util.Iterator<FlushQueueEntry> it = flushQueue.iterator();
645
646 while(it.hasNext()){
647 queueList.append(" "+it.next().toString());
648 queueList.append("\n");
649 }
650
651 return queueList.toString();
652 }
653
654
655
656
657
658 @Override
659 public void registerFlushRequestListener(final FlushRequestListener listener) {
660 this.flushRequestListeners.add(listener);
661 }
662
663
664
665
666
667
668 @Override
669 public boolean unregisterFlushRequestListener(final FlushRequestListener listener) {
670 return this.flushRequestListeners.remove(listener);
671 }
672
673
674
675
676
677 @Override
678 public void setGlobalMemstoreLimit(long globalMemStoreSize) {
679 this.globalMemStoreLimit = globalMemStoreSize;
680 this.globalMemStoreLimitLowMark =
681 (long) (this.globalMemStoreLimitLowMarkPercent * globalMemStoreSize);
682 reclaimMemStoreMemory();
683 }
684
685 public long getMemoryLimit() {
686 return this.globalMemStoreLimit;
687 }
688
689 interface FlushQueueEntry extends Delayed {
690 }
691
692
693
694
695 static class WakeupFlushThread implements FlushQueueEntry {
696 @Override
697 public long getDelay(TimeUnit unit) {
698 return 0;
699 }
700
701 @Override
702 public int compareTo(Delayed o) {
703 return -1;
704 }
705
706 @Override
707 public boolean equals(Object obj) {
708 return (this == obj);
709 }
710 }
711
712
713
714
715
716
717
718
719
720 static class FlushRegionEntry implements FlushQueueEntry {
721 private final Region region;
722
723 private final long createTime;
724 private long whenToExpire;
725 private int requeueCount = 0;
726
727 private boolean forceFlushAllStores;
728
729 FlushRegionEntry(final Region r, boolean forceFlushAllStores) {
730 this.region = r;
731 this.createTime = EnvironmentEdgeManager.currentTime();
732 this.whenToExpire = this.createTime;
733 this.forceFlushAllStores = forceFlushAllStores;
734 }
735
736
737
738
739
740 public boolean isMaximumWait(final long maximumWait) {
741 return (EnvironmentEdgeManager.currentTime() - this.createTime) > maximumWait;
742 }
743
744
745
746
747
748 public int getRequeueCount() {
749 return this.requeueCount;
750 }
751
752
753
754
755 public boolean isForceFlushAllStores() {
756 return forceFlushAllStores;
757 }
758
759
760
761
762
763
764
765 public FlushRegionEntry requeue(final long when) {
766 this.whenToExpire = EnvironmentEdgeManager.currentTime() + when;
767 this.requeueCount++;
768 return this;
769 }
770
771 @Override
772 public long getDelay(TimeUnit unit) {
773 return unit.convert(this.whenToExpire - EnvironmentEdgeManager.currentTime(),
774 TimeUnit.MILLISECONDS);
775 }
776
777 @Override
778 public int compareTo(Delayed other) {
779
780 int ret = Long.valueOf(getDelay(TimeUnit.MILLISECONDS) -
781 other.getDelay(TimeUnit.MILLISECONDS)).intValue();
782 if (ret != 0) {
783 return ret;
784 }
785 FlushQueueEntry otherEntry = (FlushQueueEntry) other;
786 return hashCode() - otherEntry.hashCode();
787 }
788
789 @Override
790 public String toString() {
791 return "[flush region "+Bytes.toStringBinary(region.getRegionInfo().getRegionName())+"]";
792 }
793
794 @Override
795 public int hashCode() {
796 int hash = (int) getDelay(TimeUnit.MILLISECONDS);
797 return hash ^ region.hashCode();
798 }
799
800 @Override
801 public boolean equals(Object obj) {
802 if (this == obj) {
803 return true;
804 }
805 if (obj == null || getClass() != obj.getClass()) {
806 return false;
807 }
808 Delayed other = (Delayed) obj;
809 return compareTo(other) == 0;
810 }
811 }
812 }