From 1c703b8674b1b19bbf712713b1648a434a017a5c Mon Sep 17 00:00:00 2001 From: "zengqiang.xu" Date: Tue, 13 Sep 2022 19:07:57 +0800 Subject: [PATCH 1/3] HDFS-16771. JN should tersely print logs about NewerTxnIdException --- .../hadoop/hdfs/qjournal/client/QuorumJournalManager.java | 3 --- .../org/apache/hadoop/hdfs/qjournal/server/Journal.java | 6 +++++- .../hadoop/hdfs/qjournal/server/JournalNodeRpcServer.java | 1 + 3 files changed, 6 insertions(+), 4 deletions(-) diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/client/QuorumJournalManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/client/QuorumJournalManager.java index 0e3a8dd0915128..b6f706faa89b96 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/client/QuorumJournalManager.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/client/QuorumJournalManager.java @@ -524,9 +524,6 @@ public void selectInputStreams(Collection streams, selectRpcInputStreams(rpcStreams, fromTxnId, onlyDurableTxns); streams.addAll(rpcStreams); return; - } catch (NewerTxnIdException ntie) { - // normal situation, we requested newer IDs than any journal has. no new streams - return; } catch (IOException ioe) { LOG.warn("Encountered exception while tailing edits >= " + fromTxnId + " via RPC; falling back to streaming.", ioe); diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/Journal.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/Journal.java index 7726377538aeb1..268f0025a30572 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/Journal.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/Journal.java @@ -751,7 +751,11 @@ public GetJournaledEditsResponseProto getJournaledEdits(long sinceTxId, "it via " + DFSConfigKeys.DFS_HA_TAILEDITS_INPROGRESS_KEY); } long highestTxId = getHighestWrittenTxId(); - if (sinceTxId > highestTxId) { + if (sinceTxId == highestTxId + 1) { + // Requested edits that don't exist yet; short-circuit the cache here + metrics.rpcEmptyResponses.incr(); + return GetJournaledEditsResponseProto.newBuilder().setTxnCount(0).build(); + } else if (sinceTxId > highestTxId + 1) { // Requested edits that don't exist yet and is newer than highestTxId. metrics.rpcEmptyResponses.incr(); throw new NewerTxnIdException( diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/JournalNodeRpcServer.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/JournalNodeRpcServer.java index ad67cf481ae70e..4e2ec16046894d 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/JournalNodeRpcServer.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/JournalNodeRpcServer.java @@ -114,6 +114,7 @@ public class JournalNodeRpcServer implements QJournalProtocol, .setVerbose(false) .build(); + this.server.addTerseExceptions(NewerTxnIdException.class); //Adding InterQJournalProtocolPB to server InterQJournalProtocolServerSideTranslatorPB From 36296305a3e576180789753c6924a22436738f8b Mon Sep 17 00:00:00 2001 From: "zengqiang.xu" Date: Thu, 15 Sep 2022 10:56:08 +0800 Subject: [PATCH 2/3] HDFS-16771. Add UT to verify the changes --- .../hadoop/hdfs/qjournal/server/Journal.java | 10 ++++- .../qjournal/server/JournalNodeRpcServer.java | 1 + .../client/TestQuorumJournalManager.java | 37 +++++++++++++++++++ 3 files changed, 46 insertions(+), 2 deletions(-) diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/Journal.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/Journal.java index 268f0025a30572..ffa613018c6a66 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/Journal.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/Journal.java @@ -752,11 +752,17 @@ public GetJournaledEditsResponseProto getJournaledEdits(long sinceTxId, } long highestTxId = getHighestWrittenTxId(); if (sinceTxId == highestTxId + 1) { - // Requested edits that don't exist yet; short-circuit the cache here + // Requested edits that don't exist yet, but this is expected, + // because namenode always get the journaled edits with the sinceTxId + // equal to image.getLastAppliedTxId() + 1. Short-circuiting the cache here + // and returning a response with a count of 0. metrics.rpcEmptyResponses.incr(); return GetJournaledEditsResponseProto.newBuilder().setTxnCount(0).build(); } else if (sinceTxId > highestTxId + 1) { - // Requested edits that don't exist yet and is newer than highestTxId. + // Requested edits that don't exist yet and this is unexpected. Means that there is a lag + // in this journal that does not contain some edits that should exist. + // Throw one NewerTxnIdException to make namenode treat this response as an exception. + // More detailed info please refer to: HDFS-16659 and HDFS-16771. metrics.rpcEmptyResponses.incr(); throw new NewerTxnIdException( "Highest txn ID available in the journal is %d, but requested txns starting at %d.", diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/JournalNodeRpcServer.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/JournalNodeRpcServer.java index 4e2ec16046894d..ab909aef2ecd84 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/JournalNodeRpcServer.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/JournalNodeRpcServer.java @@ -115,6 +115,7 @@ public class JournalNodeRpcServer implements QJournalProtocol, .build(); this.server.addTerseExceptions(NewerTxnIdException.class); + this.server.addTerseExceptions(JournaledEditsCache.CacheMissException.class); //Adding InterQJournalProtocolPB to server InterQJournalProtocolServerSideTranslatorPB diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/qjournal/client/TestQuorumJournalManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/qjournal/client/TestQuorumJournalManager.java index 84ce7c2572482f..e2ee2e365d9ffd 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/qjournal/client/TestQuorumJournalManager.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/qjournal/client/TestQuorumJournalManager.java @@ -42,8 +42,10 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.hadoop.hdfs.server.common.Util; +import org.apache.hadoop.hdfs.server.protocol.RemoteEditLogManifest; import org.apache.hadoop.net.MockDomainNameResolver; import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.ListenableFuture; import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.SettableFuture; @@ -1263,4 +1265,39 @@ private void checkRecovery(MiniJournalCluster cluster, segmentTxId); } } + + @Test + public void testSelectLatestEditsWithoutStreaming() throws Exception { + EditLogOutputStream stm = qjm.startLogSegment( + 1, NameNodeLayoutVersion.CURRENT_LAYOUT_VERSION); + // Successfully write these edits to JN0 ~ JN2 + writeTxns(stm, 1, 10); + + AtomicInteger atomicInteger = new AtomicInteger(0); + spyGetEditLogManifest(0, 11, true, atomicInteger::incrementAndGet); + spyGetEditLogManifest(1, 11, true, atomicInteger::incrementAndGet); + spyGetEditLogManifest(2, 11, true, atomicInteger::incrementAndGet); + + List streams = new ArrayList<>(); + qjm.selectInputStreams(streams, 1, true, true); + assertEquals(1, streams.size()); + assertEquals(1, streams.get(0).getFirstTxId()); + assertEquals(10, streams.get(0).getLastTxId()); + + streams.clear(); + qjm.selectInputStreams(streams, 11, true, true); + assertEquals(0, streams.size()); + assertEquals(0, atomicInteger.get()); + } + + private void spyGetEditLogManifest(int jnSpyIdx, long fromTxId, + boolean inProgressOk, Runnable preHook) { + Mockito.doAnswer((Answer>) invocation -> { + preHook.run(); + @SuppressWarnings("unchecked") + ListenableFuture result = + (ListenableFuture) invocation.callRealMethod(); + return result; + }).when(spies.get(jnSpyIdx)).getEditLogManifest(fromTxId, inProgressOk); + } } From 2aa18d06c862ed033345f82e287950b75b50e2cb Mon Sep 17 00:00:00 2001 From: "zengqiang.xu" Date: Thu, 15 Sep 2022 19:06:58 +0800 Subject: [PATCH 3/3] HDFS-16771. Remove unused import --- .../apache/hadoop/hdfs/qjournal/client/QuorumJournalManager.java | 1 - 1 file changed, 1 deletion(-) diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/client/QuorumJournalManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/client/QuorumJournalManager.java index b6f706faa89b96..faf71a7b545d91 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/client/QuorumJournalManager.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/client/QuorumJournalManager.java @@ -31,7 +31,6 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; -import org.apache.hadoop.hdfs.qjournal.server.NewerTxnIdException; import org.apache.hadoop.util.Lists; import org.slf4j.Logger; import org.slf4j.LoggerFactory;