Skip to content

Commit 23002be

Browse files
lnbest0707lnbest0707
authored
Merge branch 'apache:trunk' into YARN-11391 
2 parents e1d2a74 + 0926fa5 commit 23002be

File tree

30 files changed

+1061
-313
lines changed

30 files changed

+1061
-313
lines changed

‎hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/MembershipNamenodeResolver.java‎

Lines changed: 41 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -193,13 +193,53 @@ private void updateNameNodeState(final String nsId,
193193
}
194194
}
195195

196+
/**
197+
* Try to shuffle the multiple observer namenodes if listObserversFirst is true.
198+
* @param inputNameNodes the input FederationNamenodeContext list. If listObserversFirst is true,
199+
* all observers will be placed at the front of the collection.
200+
* @param listObserversFirst true if we need to shuffle the multiple front observer namenodes.
201+
* @return a list of FederationNamenodeContext.
202+
* @param <T> a subclass of FederationNamenodeContext.
203+
*/
204+
private <T extends FederationNamenodeContext> List<T> shuffleObserverNN(
205+
List<T> inputNameNodes, boolean listObserversFirst) {
206+
if (!listObserversFirst) {
207+
return inputNameNodes;
208+
}
209+
// Get Observers first.
210+
List<T> observerList = new ArrayList<>();
211+
for (T t : inputNameNodes) {
212+
if (t.getState() == OBSERVER) {
213+
observerList.add(t);
214+
} else {
215+
// The inputNameNodes are already sorted, so it can break
216+
// when the first non-observer is encountered.
217+
break;
218+
}
219+
}
220+
// Returns the inputNameNodes if no shuffle is required
221+
if (observerList.size() <= 1) {
222+
return inputNameNodes;
223+
}
224+
225+
// Shuffle multiple Observers
226+
Collections.shuffle(observerList);
227+
228+
List<T> ret = new ArrayList<>(inputNameNodes.size());
229+
ret.addAll(observerList);
230+
for (int i = observerList.size(); i < inputNameNodes.size(); i++) {
231+
ret.add(inputNameNodes.get(i));
232+
}
233+
return Collections.unmodifiableList(ret);
234+
}
235+
196236
@Override
197237
public List<? extends FederationNamenodeContext> getNamenodesForNameserviceId(
198238
final String nsId, boolean listObserversFirst) throws IOException {
199239

200240
List<? extends FederationNamenodeContext> ret = cacheNS.get(Pair.of(nsId, listObserversFirst));
201241
if (ret != null) {
202-
return ret;
242+
return shuffleObserverNN(ret, listObserversFirst);
203243
}
204244

205245
// Not cached, generate the value

‎hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterClientProtocol.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -702,8 +702,9 @@ public boolean truncate(String src, long newLength, String clientName)
702702
RemoteMethod method = new RemoteMethod("truncate",
703703
new Class<?>[] {String.class, long.class, String.class},
704704
new RemoteParam(), newLength, clientName);
705+
// Truncate can return true/false, so don't expect a result
705706
return rpcClient.invokeSequential(locations, method, Boolean.class,
706-
Boolean.TRUE);
707+
null);
707708
}
708709

709710
@Override

‎hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/resolver/TestNamenodeResolver.java‎

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -90,6 +90,98 @@ public void setup() throws IOException, InterruptedException {
9090
assertTrue(cleared);
9191
}
9292

93+
@Test
94+
public void testShuffleObserverNNs() throws Exception {
95+
// Add an active entry to the store
96+
NamenodeStatusReport activeReport = createNamenodeReport(
97+
NAMESERVICES[0], NAMENODES[0], HAServiceState.ACTIVE);
98+
assertTrue(namenodeResolver.registerNamenode(activeReport));
99+
100+
// Add a standby entry to the store
101+
NamenodeStatusReport standbyReport = createNamenodeReport(
102+
NAMESERVICES[0], NAMENODES[1], HAServiceState.STANDBY);
103+
assertTrue(namenodeResolver.registerNamenode(standbyReport));
104+
105+
// Load cache
106+
stateStore.refreshCaches(true);
107+
108+
// Get namenodes from state store.
109+
List<? extends FederationNamenodeContext> withoutObserver =
110+
namenodeResolver.getNamenodesForNameserviceId(NAMESERVICES[0], true);
111+
assertEquals(2, withoutObserver.size());
112+
assertEquals(FederationNamenodeServiceState.ACTIVE, withoutObserver.get(0).getState());
113+
assertEquals(FederationNamenodeServiceState.STANDBY, withoutObserver.get(1).getState());
114+
115+
// Get namenodes from cache.
116+
withoutObserver = namenodeResolver.getNamenodesForNameserviceId(NAMESERVICES[0], true);
117+
assertEquals(2, withoutObserver.size());
118+
assertEquals(FederationNamenodeServiceState.ACTIVE, withoutObserver.get(0).getState());
119+
assertEquals(FederationNamenodeServiceState.STANDBY, withoutObserver.get(1).getState());
120+
121+
// Add an observer entry to the store
122+
NamenodeStatusReport observerReport1 = createNamenodeReport(
123+
NAMESERVICES[0], NAMENODES[2], HAServiceState.OBSERVER);
124+
assertTrue(namenodeResolver.registerNamenode(observerReport1));
125+
126+
// Load cache
127+
stateStore.refreshCaches(true);
128+
129+
// Get namenodes from state store.
130+
List<? extends FederationNamenodeContext> observerList =
131+
namenodeResolver.getNamenodesForNameserviceId(NAMESERVICES[0], true);
132+
assertEquals(3, observerList.size());
133+
assertEquals(FederationNamenodeServiceState.OBSERVER, observerList.get(0).getState());
134+
assertEquals(FederationNamenodeServiceState.ACTIVE, observerList.get(1).getState());
135+
assertEquals(FederationNamenodeServiceState.STANDBY, observerList.get(2).getState());
136+
137+
// Get namenodes from cache.
138+
observerList = namenodeResolver.getNamenodesForNameserviceId(NAMESERVICES[0], true);
139+
assertEquals(3, observerList.size());
140+
assertEquals(FederationNamenodeServiceState.OBSERVER, observerList.get(0).getState());
141+
assertEquals(FederationNamenodeServiceState.ACTIVE, observerList.get(1).getState());
142+
assertEquals(FederationNamenodeServiceState.STANDBY, observerList.get(2).getState());
143+
144+
// Add one new observer entry to the store
145+
NamenodeStatusReport observerReport2 = createNamenodeReport(
146+
NAMESERVICES[0], NAMENODES[3], HAServiceState.OBSERVER);
147+
assertTrue(namenodeResolver.registerNamenode(observerReport2));
148+
149+
// Load cache
150+
stateStore.refreshCaches(true);
151+
152+
// Get namenodes from state store.
153+
List<? extends FederationNamenodeContext> observerList2 =
154+
namenodeResolver.getNamenodesForNameserviceId(NAMESERVICES[0], true);
155+
assertEquals(4, observerList2.size());
156+
assertEquals(FederationNamenodeServiceState.OBSERVER, observerList2.get(0).getState());
157+
assertEquals(FederationNamenodeServiceState.OBSERVER, observerList2.get(1).getState());
158+
assertEquals(FederationNamenodeServiceState.ACTIVE, observerList2.get(2).getState());
159+
assertEquals(FederationNamenodeServiceState.STANDBY, observerList2.get(3).getState());
160+
161+
// Get namenodes from cache.
162+
observerList2 = namenodeResolver.getNamenodesForNameserviceId(NAMESERVICES[0], true);
163+
assertEquals(4, observerList2.size());
164+
assertEquals(FederationNamenodeServiceState.OBSERVER, observerList2.get(0).getState());
165+
assertEquals(FederationNamenodeServiceState.OBSERVER, observerList2.get(1).getState());
166+
assertEquals(FederationNamenodeServiceState.ACTIVE, observerList2.get(2).getState());
167+
assertEquals(FederationNamenodeServiceState.STANDBY, observerList2.get(3).getState());
168+
169+
// Test shuffler
170+
List<? extends FederationNamenodeContext> observerList3;
171+
boolean hit = false;
172+
for (int i = 0; i < 1000; i++) {
173+
observerList3 = namenodeResolver.getNamenodesForNameserviceId(NAMESERVICES[0], true);
174+
assertEquals(FederationNamenodeServiceState.OBSERVER, observerList3.get(0).getState());
175+
assertEquals(FederationNamenodeServiceState.OBSERVER, observerList3.get(1).getState());
176+
if (observerList3.get(0).getNamenodeId().equals(observerList2.get(1).getNamenodeId()) &&
177+
observerList3.get(1).getNamenodeId().equals(observerList2.get(0).getNamenodeId())) {
178+
hit = true;
179+
break;
180+
}
181+
}
182+
assertTrue(hit);
183+
}
184+
93185
@Test
94186
public void testStateStoreDisconnected() throws Exception {
95187

‎hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterAllResolver.java‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
import org.apache.hadoop.fs.FileStatus;
3434
import org.apache.hadoop.fs.FileSystem;
3535
import org.apache.hadoop.fs.Path;
36+
import org.apache.hadoop.hdfs.DistributedFileSystem;
3637
import org.apache.hadoop.hdfs.server.federation.RouterConfigBuilder;
3738
import org.apache.hadoop.hdfs.server.federation.MiniRouterDFSCluster.NamenodeContext;
3839
import org.apache.hadoop.hdfs.server.federation.MiniRouterDFSCluster.RouterContext;
@@ -46,6 +47,7 @@
4647
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesRequest;
4748
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesResponse;
4849
import org.apache.hadoop.hdfs.server.federation.store.records.MountTable;
50+
import org.apache.hadoop.hdfs.server.namenode.TestFileTruncate;
4951
import org.junit.After;
5052
import org.junit.Before;
5153
import org.junit.Test;
@@ -191,6 +193,18 @@ private void testAll(final String path) throws Exception {
191193
assertDirsEverywhere(path, 9);
192194
assertFilesDistributed(path, 15);
193195

196+
// Test truncate
197+
String testTruncateFile = path + "/dir2/dir22/dir220/file-truncate.txt";
198+
createTestFile(routerFs, testTruncateFile);
199+
Path testTruncateFilePath = new Path(testTruncateFile);
200+
routerFs.truncate(testTruncateFilePath, 10);
201+
TestFileTruncate.checkBlockRecovery(testTruncateFilePath,
202+
(DistributedFileSystem) routerFs);
203+
assertEquals("Truncate file fails", 10,
204+
routerFs.getFileStatus(testTruncateFilePath).getLen());
205+
assertDirsEverywhere(path, 9);
206+
assertFilesDistributed(path, 16);
207+
194208
// Removing a directory should remove it from every subcluster
195209
routerFs.delete(new Path(path + "/dir2/dir22/dir220"), true);
196210
assertDirsEverywhere(path, 8);

‎hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/pom.xml‎

Lines changed: 0 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -124,26 +124,6 @@
124124
<artifactId>assertj-core</artifactId>
125125
<scope>test</scope>
126126
</dependency>
127-
<dependency>
128-
<groupId>org.junit.platform</groupId>
129-
<artifactId>junit-platform-launcher</artifactId>
130-
<scope>test</scope>
131-
</dependency>
132-
<dependency>
133-
<groupId>org.junit.jupiter</groupId>
134-
<artifactId>junit-jupiter-api</artifactId>
135-
<scope>test</scope>
136-
</dependency>
137-
<dependency>
138-
<groupId>org.junit.jupiter</groupId>
139-
<artifactId>junit-jupiter-engine</artifactId>
140-
<scope>test</scope>
141-
</dependency>
142-
<dependency>
143-
<groupId>org.junit.platform</groupId>
144-
<artifactId>junit-platform-launcher</artifactId>
145-
<scope>test</scope>
146-
</dependency>
147127
</dependencies>
148128

149129
<build>

‎hadoop-project/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@
3131

3232
<properties>
3333
<!-- Set the Release year during release -->
34-
<release-year>2022</release-year>
34+
<release-year>2023</release-year>
3535

3636
<failIfNoTests>false</failIfNoTests>
3737
<!--Whether to proceed to next module if any test failures exist-->

‎hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/state/StateMachine.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@ public interface StateMachine
2727
<STATE extends Enum<STATE>,
2828
EVENTTYPE extends Enum<EVENTTYPE>, EVENT> {
2929
public STATE getCurrentState();
30+
public STATE getPreviousState();
3031
public STATE doTransition(EVENTTYPE eventType, EVENT event)
3132
throws InvalidStateTransitionException;
3233
}

‎hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/state/StateMachineFactory.java‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -457,6 +457,7 @@ private class InternalStateMachine
457457
implements StateMachine<STATE, EVENTTYPE, EVENT> {
458458
private final OPERAND operand;
459459
private STATE currentState;
460+
private STATE previousState;
460461
private final StateTransitionListener<OPERAND, EVENT, STATE> listener;
461462

462463
InternalStateMachine(OPERAND operand, STATE initialState) {
@@ -479,14 +480,19 @@ public synchronized STATE getCurrentState() {
479480
return currentState;
480481
}
481482

483+
@Override
484+
public synchronized STATE getPreviousState() {
485+
return previousState;
486+
}
487+
482488
@Override
483489
public synchronized STATE doTransition(EVENTTYPE eventType, EVENT event)
484490
throws InvalidStateTransitionException {
485491
listener.preTransition(operand, currentState, event);
486-
STATE oldState = currentState;
492+
previousState = currentState;
487493
currentState = StateMachineFactory.this.doTransition
488494
(operand, currentState, eventType, event);
489-
listener.postTransition(operand, oldState, currentState, event);
495+
listener.postTransition(operand, previousState, currentState, event);
490496
return currentState;
491497
}
492498
}

‎hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/logaggregation/TestContainerLogsUtils.java‎

Lines changed: 14 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -38,10 +38,10 @@
3838
import org.apache.hadoop.yarn.logaggregation.filecontroller.LogAggregationFileControllerContext;
3939
import org.apache.hadoop.yarn.logaggregation.filecontroller.LogAggregationFileControllerFactory;
4040

41-
import static org.junit.jupiter.api.Assertions.assertTrue;
42-
4341
/**
4442
* This class contains several utility functions for log aggregation tests.
43+
* Any assertion libraries shouldn't be used here because this class is used by
44+
* multiple modules including MapReduce.
4545
*/
4646
public final class TestContainerLogsUtils {
4747

@@ -75,13 +75,16 @@ public static void createContainerLogFileInRemoteFS(Configuration conf,
7575
if (fs.exists(rootLogDirPath)) {
7676
fs.delete(rootLogDirPath, true);
7777
}
78-
assertTrue(fs.mkdirs(rootLogDirPath));
78+
fs.mkdirs(rootLogDirPath);
79+
// Make sure the target dir is created. If not, FileNotFoundException is thrown
80+
fs.getFileStatus(rootLogDirPath);
7981
Path appLogsDir = new Path(rootLogDirPath, appId.toString());
8082
if (fs.exists(appLogsDir)) {
8183
fs.delete(appLogsDir, true);
8284
}
83-
assertTrue(fs.mkdirs(appLogsDir));
84-
85+
fs.mkdirs(appLogsDir);
86+
// Make sure the target dir is created. If not, FileNotFoundException is thrown
87+
fs.getFileStatus(appLogsDir);
8588
createContainerLogInLocalDir(appLogsDir, containerToContent, fs, fileName);
8689
// upload container logs to remote log dir
8790

@@ -95,7 +98,9 @@ public static void createContainerLogFileInRemoteFS(Configuration conf,
9598
if (fs.exists(path) && deleteRemoteLogDir) {
9699
fs.delete(path, true);
97100
}
98-
assertTrue(fs.mkdirs(path));
101+
fs.mkdirs(path);
102+
// Make sure the target dir is created. If not, FileNotFoundException is thrown
103+
fs.getFileStatus(path);
99104
uploadContainerLogIntoRemoteDir(ugi, conf, rootLogDirList, nodeId, appId,
100105
containerToContent.keySet(), path);
101106
}
@@ -111,7 +116,9 @@ private static void createContainerLogInLocalDir(Path appLogsDir,
111116
if (fs.exists(containerLogsDir)) {
112117
fs.delete(containerLogsDir, true);
113118
}
114-
assertTrue(fs.mkdirs(containerLogsDir));
119+
fs.mkdirs(containerLogsDir);
120+
// Make sure the target dir is created. If not, FileNotFoundException is thrown
121+
fs.getFileStatus(containerLogsDir);
115122
Writer writer =
116123
new FileWriter(new File(containerLogsDir.toString(), fileName));
117124
writer.write(content);

‎hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/webapp/WebServicesTestUtils.java‎

Lines changed: 6 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -27,8 +27,7 @@
2727
import org.w3c.dom.Node;
2828
import org.w3c.dom.NodeList;
2929

30-
import static org.junit.jupiter.api.Assertions.assertEquals;
31-
import static org.junit.jupiter.api.Assertions.assertTrue;
30+
import static org.assertj.core.api.Assertions.assertThat;
3231

3332
public class WebServicesTestUtils {
3433
public static long getXmlLong(Element element, String name) {
@@ -121,28 +120,24 @@ public static String getXmlAttrString(Element element, String name) {
121120
}
122121

123122
public static void checkStringMatch(String print, String expected, String got) {
124-
assertTrue(
125-
got.matches(expected),
126-
print + " doesn't match, got: " + got + " expected: " + expected);
123+
assertThat(got).as(print).matches(expected);
127124
}
128125

129126
public static void checkStringContains(String print, String expected, String got) {
130-
assertTrue(
131-
got.contains(expected),
132-
print + " doesn't contain expected string, got: " + got + " expected: " + expected);
127+
assertThat(got).as(print).contains(expected);
133128
}
134129

135130
public static void checkStringEqual(String print, String expected, String got) {
136-
assertEquals(got, expected);
131+
assertThat(got).as(print).isEqualTo(expected);
137132
}
138133

139134
public static void assertResponseStatusCode(StatusType expected,
140135
StatusType actual) {
141-
assertResponseStatusCode(null, expected, actual);
136+
assertThat(expected.getStatusCode()).isEqualTo(actual.getStatusCode());
142137
}
143138

144139
public static void assertResponseStatusCode(String errmsg,
145140
StatusType expected, StatusType actual) {
146-
assertEquals(expected.getStatusCode(), actual.getStatusCode(), errmsg);
141+
assertThat(expected.getStatusCode()).withFailMessage(errmsg).isEqualTo(actual.getStatusCode());
147142
}
148143
}

0 commit comments

Comments
 (0)