Skip to content

HBASE-27223 Avoid data inconsistent between primary and secondary rep… #4633

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 2 commits into from
Closed

HBASE-27223 Avoid data inconsistent between primary and secondary rep… #4633

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
161d843
HBASE-27223 Avoid data inconsistent between primary and secondary rep…
comnetwork Jul 19, 2022
44f13c2
remove throws IOException
comnetwork Jul 19, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 12 additions & 6 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2863,7 +2863,7 @@ private boolean writeCanNotFlushMarkerToWAL(WriteEntry flushOpSeqIdMVCCEntry, WA
if (sink != null && !writeFlushWalMarker) {
/**
* Here for replication to secondary region replica could use {@link FlushAction#CANNOT_FLUSH}
* to recover writeFlushWalMarker is false, we create {@link WALEdit} for
* to recover when writeFlushWalMarker is false, we create {@link WALEdit} for
* {@link FlushDescriptor} and attach the {@link RegionReplicationSink#add} to the
* flushOpSeqIdMVCCEntry,see HBASE-26960 for more details.
*/
Expand Down Expand Up @@ -3694,7 +3694,7 @@ public void doPostOpCleanupForMiniBatch(
* @param familyMap Map of Cells by family
*/
protected void applyFamilyMapToMemStore(Map<byte[], List<Cell>> familyMap,
MemStoreSizing memstoreAccounting) throws IOException {
MemStoreSizing memstoreAccounting) {
for (Map.Entry<byte[], List<Cell>> e : familyMap.entrySet()) {
byte[] family = e.getKey();
List<Cell> cells = e.getValue();
Expand Down Expand Up @@ -5231,7 +5231,7 @@ public void setReadsEnabled(boolean readsEnabled) {
* scenario but that do not make sense otherwise.
*/
private void applyToMemStore(HStore store, List<Cell> cells, boolean delta,
MemStoreSizing memstoreAccounting) throws IOException {
MemStoreSizing memstoreAccounting) {
// Any change in how we update Store/MemStore needs to also be done in other applyToMemStore!!!!
boolean upsert = delta && store.getColumnFamilyDescriptor().getMaxVersions() == 1;
if (upsert) {
Expand Down Expand Up @@ -8037,11 +8037,17 @@ private WriteEntry doWALAppend(WALEdit walEdit, BatchOperation<?> batchOp,
try {
long txid = this.wal.appendData(this.getRegionInfo(), walKey, walEdit);
WriteEntry writeEntry = walKey.getWriteEntry();
this.attachRegionReplicationInWALAppend(batchOp, miniBatchOp, walKey, walEdit, writeEntry);
// Call sync on our edit.
if (txid != 0) {
sync(txid, batchOp.durability);
}
/**
* Here we attach the region replication action after the {@link HRegion#sync} is successful.
* If {@link HRegion#sync} throws exception,following
* {@link BatchOperation.writeMiniBatchOperationsToMemStore} will not be executed and we have
* no need to replicate to secondary replica.
*/
this.attachRegionReplicationInWALAppend(batchOp, miniBatchOp, walKey, walEdit, writeEntry);
return writeEntry;
} catch (IOException ioe) {
if (walKey.getWriteEntry() != null) {
Expand All @@ -8057,7 +8063,7 @@ private WriteEntry doWALAppend(WALEdit walEdit, BatchOperation<?> batchOp,
*/
private void attachRegionReplicationInWALAppend(BatchOperation<?> batchOp,
MiniBatchOperationInProgress<Mutation> miniBatchOp, WALKeyImpl walKey, WALEdit walEdit,
WriteEntry writeEntry) throws IOException {
WriteEntry writeEntry) {
if (!regionReplicationSink.isPresent()) {
return;
}
Expand Down Expand Up @@ -8086,7 +8092,7 @@ private void attachRegionReplicationInWALAppend(BatchOperation<?> batchOp,
* replica.
*/
private void doAttachReplicateRegionReplicaAction(WALKeyImpl walKey, WALEdit walEdit,
WriteEntry writeEntry) throws IOException {
WriteEntry writeEntry) {
if (walEdit == null || walEdit.isEmpty()) {
return;
}
Expand Down
3 changes: 1 addition & 2 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1880,8 +1880,7 @@ public long getSmallestReadPoint() {
* across all of them.
* @param readpoint readpoint below which we can safely remove duplicate KVs
*/
public void upsert(Iterable<Cell> cells, long readpoint, MemStoreSizing memstoreSizing)
throws IOException {
public void upsert(Iterable<Cell> cells, long readpoint, MemStoreSizing memstoreSizing) {
this.storeEngine.readLock();
try {
this.memstore.upsert(cells, readpoint, memstoreSizing);
Expand Down
2 changes: 1 addition & 1 deletion hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AbstractFSWAL.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,7 +142,7 @@ public abstract class AbstractFSWAL<W extends WriterBase> implements WAL {
"hbase.regionserver.wal.slowsync.roll.interval.ms";
protected static final int DEFAULT_SLOW_SYNC_ROLL_INTERVAL_MS = 60 * 1000; // in ms, 1 minute

protected static final String WAL_SYNC_TIMEOUT_MS = "hbase.regionserver.wal.sync.timeout";
public static final String WAL_SYNC_TIMEOUT_MS = "hbase.regionserver.wal.sync.timeout";
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is a missing part in our design, usually, if here we get a timeout exception, the only correct way is to abort the region server, as the design of WAL sync, is to succeed or die, there is no 'failure'. It is usually not a big deal is because we set a very large default value here, 5 minutes, usually the WAL system will abort the region server if it can not finish the sync within 5 minutes...

So I think we should throw a special IOException to upper layer, if we get this exception, we abort the region server.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@Apache9 ,ok, thank you very much for explanation, I would try to fix the code following your point.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this should be another issue.

Copy link
Contributor Author

@comnetwork comnetwork Jul 20, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@Apache9 , after I read the code about WAL , I have a question:
For AsyncFSWAL, basically WAL.sync could not throw any exception except that TimeoutIOException,
but for FSHLog, WAL.sync could throw any exception thrown by ProtobufLogWriter.append and
ProtobufLogWriter.sync , and when throwing these exceptions, it just requests the WAL rolling and does not abort the RegionServer, so for AsyncFSWAL,we could abort the RegionServer, but for FSHLog, it is not suitable to abort the RegionServer when WAL.sync throws the exception other than TimeoutIOException , we still need a way to avoid the situation described by this issue.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is also a problem for the FSHLog implementation. Basically, if the write to HDFS fails, we do not know whether the data has been persistent or not. The implementation for AsyncFSWAL, is to open a new writer and try to write the WAL entries again, and then adding logic in WAL split and replay to deal with duplicate entries. So for FSHLog, if it is not easy to add the same logic with AsyncFSWAL, the correct way is to abort the region server to let the failover logic to detect whether the WAL entries have been persistent or not.

Copy link
Contributor Author

@comnetwork comnetwork Jul 20, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@Apache9, ok, I think we could open two new jiras , one is for AsyncFSWAL to abort the RegionServer for TimeoutIOException, and the other is to implement the retry WAL entries logic of HLog as same as AsyncFSWAL , which is a little more complicated.

Copy link
Contributor Author

@comnetwork comnetwork Jul 21, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@Apache9 , I have opened HBASE-27230 and HBASE-27231 for these two problems.

protected static final int DEFAULT_WAL_SYNC_TIMEOUT_MS = 5 * 60 * 1000; // in ms, 5min

public static final String WAL_ROLL_MULTIPLIER = "hbase.regionserver.logroll.multiplier";
Expand Down
4 changes: 2 additions & 2 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/wal/AsyncFSWALProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,9 +62,9 @@ void init(FileSystem fs, Path path, Configuration c, boolean overwritable, long
StreamSlowMonitor monitor) throws IOException, CommonFSUtils.StreamLacksCapabilityException;
}

private EventLoopGroup eventLoopGroup;
protected EventLoopGroup eventLoopGroup;

private Class<? extends Channel> channelClass;
protected Class<? extends Channel> channelClass;

@Override
protected AsyncFSWAL createWAL() throws IOException {
Expand Down
4 changes: 4 additions & 0 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -505,4 +505,8 @@ public final WALProvider getMetaWALProvider() {
public ExcludeDatanodeManager getExcludeDatanodeManager() {
return excludeDatanodeManager;
}

public String getFactoryId() {
return this.factoryId;
}
}
Loading