HBASE-9022 TestHLogSplit.testIOEOnOutputThread fails; MORE DEBUGGING
git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@1507107 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
parent
d81ace449d
commit
260b8b644d
@ -161,7 +161,7 @@ public class HLogSplitter {
|
|||||||
128*1024*1024));
|
128*1024*1024));
|
||||||
|
|
||||||
this.minBatchSize = conf.getInt("hbase.regionserver.wal.logreplay.batch.size", 512);
|
this.minBatchSize = conf.getInt("hbase.regionserver.wal.logreplay.batch.size", 512);
|
||||||
this.distributedLogReplay = this.conf.getBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY,
|
this.distributedLogReplay = this.conf.getBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY,
|
||||||
HConstants.DEFAULT_DISTRIBUTED_LOG_REPLAY_CONFIG);
|
HConstants.DEFAULT_DISTRIBUTED_LOG_REPLAY_CONFIG);
|
||||||
|
|
||||||
this.numWriterThreads = conf.getInt("hbase.regionserver.hlog.splitlog.writer.threads", 3);
|
this.numWriterThreads = conf.getInt("hbase.regionserver.hlog.splitlog.writer.threads", 3);
|
||||||
@ -1026,12 +1026,14 @@ public class HLogSplitter {
|
|||||||
return t;
|
return t;
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
CompletionService<Void> completionService = new ExecutorCompletionService<Void>(
|
CompletionService<Void> completionService =
|
||||||
closeThreadPool);
|
new ExecutorCompletionService<Void>(closeThreadPool);
|
||||||
for (final Map.Entry<byte[], ? extends SinkWriter> writersEntry : writers.entrySet()) {
|
for (final Map.Entry<byte[], ? extends SinkWriter> writersEntry : writers.entrySet()) {
|
||||||
|
LOG.debug("Submitting close of " + ((WriterAndPath)writersEntry.getValue()).p);
|
||||||
completionService.submit(new Callable<Void>() {
|
completionService.submit(new Callable<Void>() {
|
||||||
public Void call() throws Exception {
|
public Void call() throws Exception {
|
||||||
WriterAndPath wap = (WriterAndPath) writersEntry.getValue();
|
WriterAndPath wap = (WriterAndPath) writersEntry.getValue();
|
||||||
|
LOG.debug("Closing " + wap.p);
|
||||||
try {
|
try {
|
||||||
wap.w.close();
|
wap.w.close();
|
||||||
} catch (IOException ioe) {
|
} catch (IOException ioe) {
|
||||||
@ -1039,7 +1041,7 @@ public class HLogSplitter {
|
|||||||
thrown.add(ioe);
|
thrown.add(ioe);
|
||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
LOG.info("Closed path " + wap.p + " (wrote " + wap.editsWritten + " edits in "
|
LOG.info("Closed wap " + wap.p + " (wrote " + wap.editsWritten + " edits in "
|
||||||
+ (wap.nanosSpent / 1000 / 1000) + "ms)");
|
+ (wap.nanosSpent / 1000 / 1000) + "ms)");
|
||||||
|
|
||||||
if (wap.editsWritten == 0) {
|
if (wap.editsWritten == 0) {
|
||||||
@ -1147,7 +1149,7 @@ public class HLogSplitter {
|
|||||||
thrown.add(ioe);
|
thrown.add(ioe);
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
LOG.info("Closed path " + wap.p + " (wrote " + wap.editsWritten + " edits in "
|
LOG.info("Closed log " + wap.p + " (wrote " + wap.editsWritten + " edits in "
|
||||||
+ (wap.nanosSpent / 1000 / 1000) + "ms)");
|
+ (wap.nanosSpent / 1000 / 1000) + "ms)");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -1296,19 +1298,19 @@ public class HLogSplitter {
|
|||||||
|
|
||||||
private long waitRegionOnlineTimeOut;
|
private long waitRegionOnlineTimeOut;
|
||||||
private final Set<String> recoveredRegions = Collections.synchronizedSet(new HashSet<String>());
|
private final Set<String> recoveredRegions = Collections.synchronizedSet(new HashSet<String>());
|
||||||
private final Map<String, RegionServerWriter> writers =
|
private final Map<String, RegionServerWriter> writers =
|
||||||
new ConcurrentHashMap<String, RegionServerWriter>();
|
new ConcurrentHashMap<String, RegionServerWriter>();
|
||||||
// online encoded region name -> region location map
|
// online encoded region name -> region location map
|
||||||
private final Map<String, HRegionLocation> onlineRegions =
|
private final Map<String, HRegionLocation> onlineRegions =
|
||||||
new ConcurrentHashMap<String, HRegionLocation>();
|
new ConcurrentHashMap<String, HRegionLocation>();
|
||||||
|
|
||||||
private Map<byte[], HConnection> tableNameToHConnectionMap = Collections
|
private Map<byte[], HConnection> tableNameToHConnectionMap = Collections
|
||||||
.synchronizedMap(new TreeMap<byte[], HConnection>(Bytes.BYTES_COMPARATOR));
|
.synchronizedMap(new TreeMap<byte[], HConnection>(Bytes.BYTES_COMPARATOR));
|
||||||
/**
|
/**
|
||||||
* Map key -> value layout
|
* Map key -> value layout
|
||||||
* <servername>:<table name> -> Queue<Row>
|
* <servername>:<table name> -> Queue<Row>
|
||||||
*/
|
*/
|
||||||
private Map<String, List<Pair<HRegionLocation, Row>>> serverToBufferQueueMap =
|
private Map<String, List<Pair<HRegionLocation, Row>>> serverToBufferQueueMap =
|
||||||
new ConcurrentHashMap<String, List<Pair<HRegionLocation, Row>>>();
|
new ConcurrentHashMap<String, List<Pair<HRegionLocation, Row>>>();
|
||||||
private List<Throwable> thrown = new ArrayList<Throwable>();
|
private List<Throwable> thrown = new ArrayList<Throwable>();
|
||||||
|
|
||||||
@ -1321,7 +1323,7 @@ public class HLogSplitter {
|
|||||||
|
|
||||||
public LogReplayOutputSink(int numWriters) {
|
public LogReplayOutputSink(int numWriters) {
|
||||||
super(numWriters);
|
super(numWriters);
|
||||||
this.waitRegionOnlineTimeOut = conf.getInt("hbase.splitlog.manager.timeout",
|
this.waitRegionOnlineTimeOut = conf.getInt("hbase.splitlog.manager.timeout",
|
||||||
SplitLogManager.DEFAULT_TIMEOUT);
|
SplitLogManager.DEFAULT_TIMEOUT);
|
||||||
this.logRecoveredEditsOutputSink = new LogRecoveredEditsOutputSink(numWriters);
|
this.logRecoveredEditsOutputSink = new LogRecoveredEditsOutputSink(numWriters);
|
||||||
this.logRecoveredEditsOutputSink.setReporter(reporter);
|
this.logRecoveredEditsOutputSink.setReporter(reporter);
|
||||||
@ -1333,7 +1335,7 @@ public class HLogSplitter {
|
|||||||
LOG.warn("got an empty buffer, skipping");
|
LOG.warn("got an empty buffer, skipping");
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
// check if current region in a disabling or disabled table
|
// check if current region in a disabling or disabled table
|
||||||
if (disablingOrDisabledTables.contains(Bytes.toString(buffer.tableName))) {
|
if (disablingOrDisabledTables.contains(Bytes.toString(buffer.tableName))) {
|
||||||
// need fall back to old way
|
// need fall back to old way
|
||||||
@ -1433,7 +1435,7 @@ public class HLogSplitter {
|
|||||||
}
|
}
|
||||||
|
|
||||||
try {
|
try {
|
||||||
loc = locateRegionAndRefreshLastFlushedSequenceId(hconn, table, kv.getRow(),
|
loc = locateRegionAndRefreshLastFlushedSequenceId(hconn, table, kv.getRow(),
|
||||||
encodeRegionNameStr);
|
encodeRegionNameStr);
|
||||||
} catch (TableNotFoundException ex) {
|
} catch (TableNotFoundException ex) {
|
||||||
// table has been deleted so skip edits of the table
|
// table has been deleted so skip edits of the table
|
||||||
@ -1493,7 +1495,7 @@ public class HLogSplitter {
|
|||||||
|
|
||||||
// skip the edit
|
// skip the edit
|
||||||
if(needSkip) continue;
|
if(needSkip) continue;
|
||||||
|
|
||||||
// add the last row
|
// add the last row
|
||||||
if (preRow != null && lastAddedRow != preRow) {
|
if (preRow != null && lastAddedRow != preRow) {
|
||||||
synchronized (serverToBufferQueueMap) {
|
synchronized (serverToBufferQueueMap) {
|
||||||
@ -1554,7 +1556,7 @@ public class HLogSplitter {
|
|||||||
}
|
}
|
||||||
regionMaxSeqIdInStores.put(loc.getRegionInfo().getEncodedName(), storeIds);
|
regionMaxSeqIdInStores.put(loc.getRegionInfo().getEncodedName(), storeIds);
|
||||||
}
|
}
|
||||||
|
|
||||||
if (cachedLastFlushedSequenceId == null
|
if (cachedLastFlushedSequenceId == null
|
||||||
|| lastFlushedSequenceId > cachedLastFlushedSequenceId) {
|
|| lastFlushedSequenceId > cachedLastFlushedSequenceId) {
|
||||||
lastFlushedSequenceIds.put(loc.getRegionInfo().getEncodedName(), lastFlushedSequenceId);
|
lastFlushedSequenceIds.put(loc.getRegionInfo().getEncodedName(), lastFlushedSequenceId);
|
||||||
@ -1599,7 +1601,7 @@ public class HLogSplitter {
|
|||||||
*/
|
*/
|
||||||
private HRegionLocation waitUntilRegionOnline(HRegionLocation loc, byte[] row,
|
private HRegionLocation waitUntilRegionOnline(HRegionLocation loc, byte[] row,
|
||||||
final long timeout)
|
final long timeout)
|
||||||
throws IOException {
|
throws IOException {
|
||||||
final long endTime = EnvironmentEdgeManager.currentTimeMillis() + timeout;
|
final long endTime = EnvironmentEdgeManager.currentTimeMillis() + timeout;
|
||||||
final long pause = conf.getLong(HConstants.HBASE_CLIENT_PAUSE,
|
final long pause = conf.getLong(HConstants.HBASE_CLIENT_PAUSE,
|
||||||
HConstants.DEFAULT_HBASE_CLIENT_PAUSE);
|
HConstants.DEFAULT_HBASE_CLIENT_PAUSE);
|
||||||
@ -1631,12 +1633,12 @@ public class HLogSplitter {
|
|||||||
Thread.sleep(expectedSleep);
|
Thread.sleep(expectedSleep);
|
||||||
} catch (InterruptedException e) {
|
} catch (InterruptedException e) {
|
||||||
Thread.currentThread().interrupt();
|
Thread.currentThread().interrupt();
|
||||||
throw new IOException("Interrupted when waiting regon " +
|
throw new IOException("Interrupted when waiting regon " +
|
||||||
loc.getRegionInfo().getEncodedName() + " online.", e);
|
loc.getRegionInfo().getEncodedName() + " online.", e);
|
||||||
}
|
}
|
||||||
tries++;
|
tries++;
|
||||||
}
|
}
|
||||||
|
|
||||||
throw new IOException("Timeout when waiting region " + loc.getRegionInfo().getEncodedName() +
|
throw new IOException("Timeout when waiting region " + loc.getRegionInfo().getEncodedName() +
|
||||||
" online for " + timeout + " milliseconds.", cause);
|
" online for " + timeout + " milliseconds.", cause);
|
||||||
}
|
}
|
||||||
@ -1802,7 +1804,7 @@ public class HLogSplitter {
|
|||||||
}
|
}
|
||||||
return hconn;
|
return hconn;
|
||||||
}
|
}
|
||||||
|
|
||||||
private String getTableFromLocationStr(String loc) {
|
private String getTableFromLocationStr(String loc) {
|
||||||
/**
|
/**
|
||||||
* location key is in format <server name:port>#<table name>
|
* location key is in format <server name:port>#<table name>
|
||||||
|
Loading…
x
Reference in New Issue
Block a user