diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/FSHLog.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/FSHLog.java index 268e302..939d1e7 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/FSHLog.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/FSHLog.java @@ -1178,13 +1178,21 @@ class FSHLog implements HLog, Syncable { // LinkedBlockingQueue because of // http://www.javacodegeeks.com/2010/09/java-best-practices-queue-battle-and.html // Could use other blockingqueues here or concurrent queues. - this.syncFutures = new LinkedBlockingQueue(maxHandlersCount); + // Multiply by 2 so excess capacity just in case. HBASE-11145 has us able to get full. + // Is HBASE-11145 an off-by-one? Something is wrong if we are this backed up. + this.syncFutures = new LinkedBlockingQueue(maxHandlersCount * 2); } - void offer(final long sequence, final SyncFuture [] syncFutures, final int syncFutureCount) { + void offer(final long sequence, final SyncFuture [] syncFutures, final int syncFutureCount) + throws InterruptedException { // Set sequence first because the add to the queue will wake the thread if sleeping. this.sequence = sequence; - for (int i = 0; i < syncFutureCount; i++) this.syncFutures.add(syncFutures[i]); + for (int i = 0; i < syncFutureCount; i++) { + while (!this.syncFutures.offer(syncFutures[i], 1000, TimeUnit.MILLISECONDS)) { + LOG.warn("Sync queue full, retrying; size=" + this.syncFutures.size() + + ", remainingCapacity=" + this.syncFutures.remainingCapacity()); + } + } } /** @@ -1295,7 +1303,7 @@ class FSHLog implements HLog, Syncable { // Presume legit interrupt. Thread.currentThread().interrupt(); } catch (Throwable t) { - LOG.warn("UNEXPECTED, continuing", t); + LOG.error("UNEXPECTED", t); } } }