Skip to content

Commit 1a8dbf3

Browse files
tomscuttomscut
committed
HDFS-16331. Make dfs.blockreport.intervalMsec reconfigurable
1 parent 91af256 commit 1a8dbf3

File tree

4 files changed

+176
-59
lines changed

4 files changed

+176
-59
lines changed

‎hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BPServiceActor.java‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1190,7 +1190,7 @@ static class Scheduler {
11901190

11911191
private final long heartbeatIntervalMs;
11921192
private final long lifelineIntervalMs;
1193-
private final long blockReportIntervalMs;
1193+
private volatile long blockReportIntervalMs;
11941194
private final long outliersReportIntervalMs;
11951195

11961196
Scheduler(long heartbeatIntervalMs, long lifelineIntervalMs,
@@ -1346,6 +1346,14 @@ void setNextBlockReportTime(long nextBlockReportTime) {
13461346
this.nextBlockReportTime.getAndSet(nextBlockReportTime);
13471347
}
13481348

1349+
long getBlockReportIntervalMs() {
1350+
return this.blockReportIntervalMs;
1351+
}
1352+
1353+
void setBlockReportIntervalMs(long intervalMs) {
1354+
this.blockReportIntervalMs = intervalMs;
1355+
}
1356+
13491357
/**
13501358
* Wrapped for testing.
13511359
* @return

‎hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java‎

Lines changed: 100 additions & 57 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,8 @@
1818
package org.apache.hadoop.hdfs.server.datanode;
1919

2020

21+
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_BLOCKREPORT_INTERVAL_MSEC_KEY;
22+
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_BLOCKREPORT_INTERVAL_MSEC_DEFAULT;
2123
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_ADDRESS_DEFAULT;
2224
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_ADDRESS_KEY;
2325
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_ALLOW_SAME_DISK_TIERING;
@@ -299,7 +301,8 @@ public class DataNode extends ReconfigurableBase
299301
Collections.unmodifiableList(
300302
Arrays.asList(
301303
DFS_DATANODE_DATA_DIR_KEY,
302-
DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_KEY));
304+
DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_KEY,
305+
DFS_BLOCKREPORT_INTERVAL_MSEC_KEY));
303306

304307
public static final Log METRICS_LOG = LogFactory.getLog("DataNodeMetricsLog");
305308

@@ -533,78 +536,118 @@ protected Configuration getNewConf() {
533536
public String reconfigurePropertyImpl(String property, String newVal)
534537
throws ReconfigurationException {
535538
switch (property) {
536-
case DFS_DATANODE_DATA_DIR_KEY: {
537-
IOException rootException = null;
539+
case DFS_DATANODE_DATA_DIR_KEY: {
540+
IOException rootException = null;
541+
try {
542+
LOG.info("Reconfiguring {} to {}", property, newVal);
543+
this.refreshVolumes(newVal);
544+
return getConf().get(DFS_DATANODE_DATA_DIR_KEY);
545+
} catch (IOException e) {
546+
rootException = e;
547+
} finally {
548+
// Send a full block report to let NN acknowledge the volume changes.
538549
try {
539-
LOG.info("Reconfiguring {} to {}", property, newVal);
540-
this.refreshVolumes(newVal);
541-
return getConf().get(DFS_DATANODE_DATA_DIR_KEY);
550+
triggerBlockReport(
551+
new BlockReportOptions.Factory().setIncremental(false).build());
542552
} catch (IOException e) {
543-
rootException = e;
553+
LOG.warn("Exception while sending the block report after refreshing"
554+
+ " volumes {} to {}", property, newVal, e);
555+
if (rootException == null) {
556+
rootException = e;
557+
}
544558
} finally {
545-
// Send a full block report to let NN acknowledge the volume changes.
546-
try {
547-
triggerBlockReport(
548-
new BlockReportOptions.Factory().setIncremental(false).build());
549-
} catch (IOException e) {
550-
LOG.warn("Exception while sending the block report after refreshing"
551-
+ " volumes {} to {}", property, newVal, e);
552-
if (rootException == null) {
553-
rootException = e;
554-
}
555-
} finally {
556-
if (rootException != null) {
557-
throw new ReconfigurationException(property, newVal,
558-
getConf().get(property), rootException);
559-
}
559+
if (rootException != null) {
560+
throw new ReconfigurationException(property, newVal,
561+
getConf().get(property), rootException);
560562
}
561563
}
562-
break;
563564
}
564-
case DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_KEY: {
565-
ReconfigurationException rootException = null;
566-
try {
567-
LOG.info("Reconfiguring {} to {}", property, newVal);
568-
int movers;
569-
if (newVal == null) {
570-
// set to default
571-
movers = DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_DEFAULT;
572-
} else {
573-
movers = Integer.parseInt(newVal);
574-
if (movers <= 0) {
575-
rootException = new ReconfigurationException(
576-
property,
577-
newVal,
578-
getConf().get(property),
579-
new IllegalArgumentException(
580-
"balancer max concurrent movers must be larger than 0"));
581-
}
582-
}
583-
boolean success = xserver.updateBalancerMaxConcurrentMovers(movers);
584-
if (!success) {
565+
break;
566+
}
567+
case DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_KEY: {
568+
ReconfigurationException rootException = null;
569+
try {
570+
LOG.info("Reconfiguring {} to {}", property, newVal);
571+
int movers;
572+
if (newVal == null) {
573+
// set to default
574+
movers = DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_DEFAULT;
575+
} else {
576+
movers = Integer.parseInt(newVal);
577+
if (movers <= 0) {
585578
rootException = new ReconfigurationException(
586579
property,
587580
newVal,
588581
getConf().get(property),
589582
new IllegalArgumentException(
590-
"Could not modify concurrent moves thread count"));
583+
"balancer max concurrent movers must be larger than 0"));
591584
}
592-
return Integer.toString(movers);
593-
} catch (NumberFormatException nfe) {
585+
}
586+
boolean success = xserver.updateBalancerMaxConcurrentMovers(movers);
587+
if (!success) {
594588
rootException = new ReconfigurationException(
595-
property, newVal, getConf().get(property), nfe);
596-
} finally {
597-
if (rootException != null) {
598-
LOG.warn(String.format(
599-
"Exception in updating balancer max concurrent movers %s to %s",
600-
property, newVal), rootException);
601-
throw rootException;
589+
property,
590+
newVal,
591+
getConf().get(property),
592+
new IllegalArgumentException(
593+
"Could not modify concurrent moves thread count"));
594+
}
595+
return Integer.toString(movers);
596+
} catch (NumberFormatException nfe) {
597+
rootException = new ReconfigurationException(
598+
property, newVal, getConf().get(property), nfe);
599+
} finally {
600+
if (rootException != null) {
601+
LOG.warn(String.format(
602+
"Exception in updating balancer max concurrent movers %s to %s",
603+
property, newVal), rootException);
604+
throw rootException;
605+
}
606+
}
607+
break;
608+
}
609+
case DFS_BLOCKREPORT_INTERVAL_MSEC_KEY: {
610+
ReconfigurationException rootException = null;
611+
try {
612+
LOG.info("Reconfiguring {} to {}", property, newVal);
613+
long intervalMs;
614+
if (newVal == null) {
615+
// Set to default.
616+
intervalMs = DFS_BLOCKREPORT_INTERVAL_MSEC_DEFAULT;
617+
} else {
618+
intervalMs = Long.parseLong(newVal);
619+
if (intervalMs < 0) {
620+
rootException = new ReconfigurationException(
621+
property,
622+
newVal,
623+
getConf().get(property),
624+
new IllegalArgumentException(
625+
"block report interval must be larger than or equal to 0"));
626+
}
627+
}
628+
for (BPOfferService bpos : blockPoolManager.getAllNamenodeThreads()) {
629+
if (bpos != null) {
630+
for (BPServiceActor actor : bpos.getBPServiceActors()) {
631+
actor.getScheduler().setBlockReportIntervalMs(intervalMs);
632+
}
602633
}
603634
}
604-
break;
635+
return Long.toString(intervalMs);
636+
} catch (NumberFormatException nfe) {
637+
rootException = new ReconfigurationException(
638+
property, newVal, getConf().get(property), nfe);
639+
} finally {
640+
if (rootException != null) {
641+
LOG.warn(String.format(
642+
"Exception in updating block report interval %s to %s",
643+
property, newVal), rootException);
644+
throw rootException;
645+
}
605646
}
606-
default:
607-
break;
647+
break;
648+
}
649+
default:
650+
break;
608651
}
609652
throw new ReconfigurationException(
610653
property, newVal, getConf().get(property));

‎hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeReconfiguration.java‎

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,8 @@
1818

1919
package org.apache.hadoop.hdfs.server.datanode;
2020

21+
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_BLOCKREPORT_INTERVAL_MSEC_DEFAULT;
22+
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_BLOCKREPORT_INTERVAL_MSEC_KEY;
2123
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_KEY;
2224
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_DEFAULT;
2325
import static org.junit.Assert.assertEquals;
@@ -293,4 +295,68 @@ private void testAcquireOnMaxConcurrentMoversReconfiguration(
293295
assertEquals("should not be able to get thread quota", false,
294296
dataNode.xserver.balanceThrottler.acquire());
295297
}
298+
299+
@Test
300+
public void testBlockReportIntervalReconfiguration()
301+
throws ReconfigurationException {
302+
int blockReportInterval = 300 * 1000;
303+
for (int i = 0; i < NUM_DATA_NODE; i++) {
304+
DataNode dn = cluster.getDataNodes().get(i);
305+
306+
// Try invalid values.
307+
try {
308+
dn.reconfigureProperty(
309+
DFS_BLOCKREPORT_INTERVAL_MSEC_KEY, "text");
310+
fail("ReconfigurationException expected");
311+
} catch (ReconfigurationException expected) {
312+
assertTrue("expecting NumberFormatException",
313+
expected.getCause() instanceof NumberFormatException);
314+
}
315+
try {
316+
dn.reconfigureProperty(
317+
DFS_BLOCKREPORT_INTERVAL_MSEC_KEY,
318+
String.valueOf(-1));
319+
fail("ReconfigurationException expected");
320+
} catch (ReconfigurationException expected) {
321+
assertTrue("expecting IllegalArgumentException",
322+
expected.getCause() instanceof IllegalArgumentException);
323+
}
324+
325+
// Change properties.
326+
dn.reconfigureProperty(DFS_BLOCKREPORT_INTERVAL_MSEC_KEY,
327+
String.valueOf(blockReportInterval));
328+
329+
// Verify change.
330+
BlockPoolManager blockPoolManager = new BlockPoolManager(dn);
331+
for (BPOfferService bpos : blockPoolManager.getAllNamenodeThreads()) {
332+
if (bpos != null) {
333+
for (BPServiceActor actor : bpos.getBPServiceActors()) {
334+
assertEquals(String.format("%s has wrong value",
335+
DFS_BLOCKREPORT_INTERVAL_MSEC_KEY),
336+
blockReportInterval,
337+
actor.getScheduler().getBlockReportIntervalMs());
338+
}
339+
}
340+
}
341+
342+
// Revert to default.
343+
dn.reconfigureProperty(DFS_BLOCKREPORT_INTERVAL_MSEC_KEY,
344+
null);
345+
346+
// Verify default.
347+
for (BPOfferService bpos : blockPoolManager.getAllNamenodeThreads()) {
348+
if (bpos != null) {
349+
for (BPServiceActor actor : bpos.getBPServiceActors()) {
350+
assertEquals(String.format("%s has wrong value",
351+
DFS_BLOCKREPORT_INTERVAL_MSEC_KEY),
352+
DFS_BLOCKREPORT_INTERVAL_MSEC_DEFAULT,
353+
actor.getScheduler().getBlockReportIntervalMs());
354+
}
355+
}
356+
}
357+
assertEquals(String.format("expect %s is not configured",
358+
DFS_BLOCKREPORT_INTERVAL_MSEC_KEY), null, dn
359+
.getConf().get(DFS_BLOCKREPORT_INTERVAL_MSEC_KEY));
360+
}
361+
}
296362
}

‎hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/TestDFSAdmin.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -336,7 +336,7 @@ public void testDataNodeGetReconfigurableProperties() throws IOException {
336336
final List<String> outs = Lists.newArrayList();
337337
final List<String> errs = Lists.newArrayList();
338338
getReconfigurableProperties("datanode", address, outs, errs);
339-
assertEquals(3, outs.size());
339+
assertEquals(4, outs.size());
340340
assertEquals(DFSConfigKeys.DFS_DATANODE_DATA_DIR_KEY, outs.get(1));
341341
}
342342

0 commit comments

Comments
 (0)