JeongDaeKim commented on a change in pull request #749: HBASE-23205 Correctly
update the position of WALs currently being replicated
URL: https://github.com/apache/hbase/pull/749#discussion_r348419504
##########
File path:
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReaderThread.java
##########
@@ -135,59 +127,46 @@ public void run() {
try (WALEntryStream entryStream =
new WALEntryStream(logQueue, fs, conf, currentPosition, metrics)) {
while (isReaderRunning()) { // loop here to keep reusing stream while
we can
- if (!checkQuota()) {
+ if (manager.isBufferQuotaReached()) {
+ Threads.sleep(sleepForRetries);
continue;
}
- WALEntryBatch batch = null;
- while (entryStream.hasNext()) {
- if (batch == null) {
- batch = new WALEntryBatch(replicationBatchCountCapacity,
entryStream.getCurrentPath());
- }
+ WALEntryBatch batch =
+ new WALEntryBatch(replicationBatchCountCapacity,
replicationBatchSizeCapacity);
+ boolean hasNext;
+ while ((hasNext = entryStream.hasNext()) == true) {
Entry entry = entryStream.next();
entry = filterEntry(entry);
if (entry != null) {
WALEdit edit = entry.getEdit();
if (edit != null && !edit.isEmpty()) {
- long entrySize = getEntrySizeIncludeBulkLoad(entry);
- long entrySizeExlucdeBulkLoad =
getEntrySizeExcludeBulkLoad(entry);
- batch.addEntry(entry);
- replicationSourceManager.setPendingShipment(true);
- updateBatchStats(batch, entry, entryStream.getPosition(),
entrySize);
- boolean totalBufferTooLarge =
acquireBufferQuota(entrySizeExlucdeBulkLoad);
+ long entrySizeExcludeBulkLoad = batch.addEntry(entry);
+ boolean totalBufferTooLarge =
manager.acquireBufferQuota(entrySizeExcludeBulkLoad);
// Stop if too many entries or too big
- if (totalBufferTooLarge || batch.getHeapSize() >=
replicationBatchSizeCapacity
- || batch.getNbEntries() >= replicationBatchCountCapacity) {
+ if (totalBufferTooLarge || batch.isLimitReached()) {
break;
}
}
- } else {
Review comment:
> What if the whole WAL section read got no entries for replication? In this
case, batch would be empty, so
_ReplicationSourceManager.logPositionAndCleanOldLogs_ does not ever get called
(at least, I guess, until the log is rolled).
Yes, In that case, the position will be updated when log rolled. That is my
intention. https://github.com/apache/hbase/pull/749#discussion_r341050152
> I think the answer to my question above is in the resetStream() that gets
called at the end of the second while loop, which will update lastReadPosition
variable that is now used for reading here.
Oh? Then, I think i didn't understand what your question is. 🤣
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
