Skip to content

Commit cdaf4d8

Browse files
tomscuttasanuma
tomscut
authored and
tasanuma
committed
HDFS-16331. Make dfs.blockreport.intervalMsec reconfigurable (#3676)
Signed-off-by: Takanobu Asanuma <[email protected]> (cherry picked from commit 52ec65f)
1 parent 831c11c commit cdaf4d8

File tree

5 files changed

+188
-60
lines changed

5 files changed

+188
-60
lines changed

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

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,7 @@
6767
import org.apache.hadoop.io.IOUtils;
6868
import org.apache.hadoop.ipc.RemoteException;
6969
import org.apache.hadoop.net.NetUtils;
70+
import org.apache.hadoop.util.Preconditions;
7071
import org.apache.hadoop.util.Time;
7172
import org.apache.hadoop.util.VersionInfo;
7273
import org.apache.hadoop.util.VersionUtil;
@@ -1139,7 +1140,7 @@ static class Scheduler {
11391140

11401141
private final long heartbeatIntervalMs;
11411142
private final long lifelineIntervalMs;
1142-
private final long blockReportIntervalMs;
1143+
private volatile long blockReportIntervalMs;
11431144
private final long outliersReportIntervalMs;
11441145

11451146
Scheduler(long heartbeatIntervalMs, long lifelineIntervalMs,
@@ -1296,6 +1297,15 @@ void setNextBlockReportTime(long nextBlockReportTime) {
12961297
this.nextBlockReportTime.getAndSet(nextBlockReportTime);
12971298
}
12981299

1300+
long getBlockReportIntervalMs() {
1301+
return this.blockReportIntervalMs;
1302+
}
1303+
1304+
void setBlockReportIntervalMs(long intervalMs) {
1305+
Preconditions.checkArgument(intervalMs > 0);
1306+
this.blockReportIntervalMs = intervalMs;
1307+
}
1308+
12991309
/**
13001310
* Wrapped for testing.
13011311
* @return

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

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -74,6 +74,7 @@
7474
import org.apache.hadoop.hdfs.protocol.datatransfer.sasl.DataTransferSaslUtil;
7575
import org.apache.hadoop.hdfs.server.common.Util;
7676
import org.apache.hadoop.security.SaslPropertiesResolver;
77+
import org.apache.hadoop.util.Preconditions;
7778

7879
import java.util.concurrent.TimeUnit;
7980

@@ -105,7 +106,7 @@ public class DNConf {
105106
final long readaheadLength;
106107
final long heartBeatInterval;
107108
private final long lifelineIntervalMs;
108-
final long blockReportInterval;
109+
volatile long blockReportInterval;
109110
final long blockReportSplitThreshold;
110111
final boolean peerStatsEnabled;
111112
final boolean diskStatsEnabled;
@@ -474,4 +475,13 @@ public boolean getPmemCacheRecoveryEnabled() {
474475
public long getProcessCommandsThresholdMs() {
475476
return processCommandsThresholdMs;
476477
}
478+
479+
void setBlockReportInterval(long intervalMs) {
480+
Preconditions.checkArgument(intervalMs > 0);
481+
blockReportInterval = intervalMs;
482+
}
483+
484+
public long getBlockReportInterval() {
485+
return blockReportInterval;
486+
}
477487
}

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

Lines changed: 93 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_DATA_DIR_KEY;
@@ -304,7 +306,8 @@ public class DataNode extends ReconfigurableBase
304306
Collections.unmodifiableList(
305307
Arrays.asList(
306308
DFS_DATANODE_DATA_DIR_KEY,
307-
DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_KEY));
309+
DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_KEY,
310+
DFS_BLOCKREPORT_INTERVAL_MSEC_KEY));
308311

309312
public static final Log METRICS_LOG = LogFactory.getLog("DataNodeMetricsLog");
310313

@@ -540,78 +543,111 @@ protected Configuration getNewConf() {
540543
public String reconfigurePropertyImpl(String property, String newVal)
541544
throws ReconfigurationException {
542545
switch (property) {
543-
case DFS_DATANODE_DATA_DIR_KEY: {
544-
IOException rootException = null;
546+
case DFS_DATANODE_DATA_DIR_KEY: {
547+
IOException rootException = null;
548+
try {
549+
LOG.info("Reconfiguring {} to {}", property, newVal);
550+
this.refreshVolumes(newVal);
551+
return getConf().get(DFS_DATANODE_DATA_DIR_KEY);
552+
} catch (IOException e) {
553+
rootException = e;
554+
} finally {
555+
// Send a full block report to let NN acknowledge the volume changes.
545556
try {
546-
LOG.info("Reconfiguring {} to {}", property, newVal);
547-
this.refreshVolumes(newVal);
548-
return getConf().get(DFS_DATANODE_DATA_DIR_KEY);
557+
triggerBlockReport(
558+
new BlockReportOptions.Factory().setIncremental(false).build());
549559
} catch (IOException e) {
550-
rootException = e;
560+
LOG.warn("Exception while sending the block report after refreshing"
561+
+ " volumes {} to {}", property, newVal, e);
562+
if (rootException == null) {
563+
rootException = e;
564+
}
551565
} finally {
552-
// Send a full block report to let NN acknowledge the volume changes.
553-
try {
554-
triggerBlockReport(
555-
new BlockReportOptions.Factory().setIncremental(false).build());
556-
} catch (IOException e) {
557-
LOG.warn("Exception while sending the block report after refreshing"
558-
+ " volumes {} to {}", property, newVal, e);
559-
if (rootException == null) {
560-
rootException = e;
561-
}
562-
} finally {
563-
if (rootException != null) {
564-
throw new ReconfigurationException(property, newVal,
565-
getConf().get(property), rootException);
566-
}
566+
if (rootException != null) {
567+
throw new ReconfigurationException(property, newVal,
568+
getConf().get(property), rootException);
567569
}
568570
}
569-
break;
570571
}
571-
case DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_KEY: {
572-
ReconfigurationException rootException = null;
573-
try {
574-
LOG.info("Reconfiguring {} to {}", property, newVal);
575-
int movers;
576-
if (newVal == null) {
577-
// set to default
578-
movers = DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_DEFAULT;
579-
} else {
580-
movers = Integer.parseInt(newVal);
581-
if (movers <= 0) {
582-
rootException = new ReconfigurationException(
583-
property,
584-
newVal,
585-
getConf().get(property),
586-
new IllegalArgumentException(
587-
"balancer max concurrent movers must be larger than 0"));
588-
}
589-
}
590-
boolean success = xserver.updateBalancerMaxConcurrentMovers(movers);
591-
if (!success) {
572+
break;
573+
}
574+
case DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_KEY: {
575+
ReconfigurationException rootException = null;
576+
try {
577+
LOG.info("Reconfiguring {} to {}", property, newVal);
578+
int movers;
579+
if (newVal == null) {
580+
// set to default
581+
movers = DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_DEFAULT;
582+
} else {
583+
movers = Integer.parseInt(newVal);
584+
if (movers <= 0) {
592585
rootException = new ReconfigurationException(
593586
property,
594587
newVal,
595588
getConf().get(property),
596589
new IllegalArgumentException(
597-
"Could not modify concurrent moves thread count"));
590+
"balancer max concurrent movers must be larger than 0"));
598591
}
599-
return Integer.toString(movers);
600-
} catch (NumberFormatException nfe) {
592+
}
593+
boolean success = xserver.updateBalancerMaxConcurrentMovers(movers);
594+
if (!success) {
601595
rootException = new ReconfigurationException(
602-
property, newVal, getConf().get(property), nfe);
603-
} finally {
604-
if (rootException != null) {
605-
LOG.warn(String.format(
606-
"Exception in updating balancer max concurrent movers %s to %s",
607-
property, newVal), rootException);
608-
throw rootException;
596+
property,
597+
newVal,
598+
getConf().get(property),
599+
new IllegalArgumentException(
600+
"Could not modify concurrent moves thread count"));
601+
}
602+
return Integer.toString(movers);
603+
} catch (NumberFormatException nfe) {
604+
rootException = new ReconfigurationException(
605+
property, newVal, getConf().get(property), nfe);
606+
} finally {
607+
if (rootException != null) {
608+
LOG.warn(String.format(
609+
"Exception in updating balancer max concurrent movers %s to %s",
610+
property, newVal), rootException);
611+
throw rootException;
612+
}
613+
}
614+
break;
615+
}
616+
case DFS_BLOCKREPORT_INTERVAL_MSEC_KEY: {
617+
ReconfigurationException rootException = null;
618+
try {
619+
LOG.info("Reconfiguring {} to {}", property, newVal);
620+
long intervalMs;
621+
if (newVal == null) {
622+
// Set to default.
623+
intervalMs = DFS_BLOCKREPORT_INTERVAL_MSEC_DEFAULT;
624+
} else {
625+
intervalMs = Long.parseLong(newVal);
626+
}
627+
dnConf.setBlockReportInterval(intervalMs);
628+
for (BPOfferService bpos : blockPoolManager.getAllNamenodeThreads()) {
629+
if (bpos != null) {
630+
for (BPServiceActor actor : bpos.getBPServiceActors()) {
631+
actor.getScheduler().setBlockReportIntervalMs(intervalMs);
632+
}
609633
}
610634
}
611-
break;
635+
return Long.toString(intervalMs);
636+
} catch (IllegalArgumentException e) {
637+
rootException = new ReconfigurationException(
638+
property, newVal, getConf().get(property), e);
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+
}
612646
}
613-
default:
614-
break;
647+
break;
648+
}
649+
default:
650+
break;
615651
}
616652
throw new ReconfigurationException(
617653
property, newVal, getConf().get(property));

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

Lines changed: 72 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,74 @@ 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, IOException {
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+
assertEquals(String.format("%s has wrong value",
331+
DFS_BLOCKREPORT_INTERVAL_MSEC_KEY),
332+
blockReportInterval,
333+
dn.getDnConf().getBlockReportInterval());
334+
for (BPOfferService bpos : dn.getAllBpOs()) {
335+
if (bpos != null) {
336+
for (BPServiceActor actor : bpos.getBPServiceActors()) {
337+
assertEquals(String.format("%s has wrong value",
338+
DFS_BLOCKREPORT_INTERVAL_MSEC_KEY),
339+
blockReportInterval,
340+
actor.getScheduler().getBlockReportIntervalMs());
341+
}
342+
}
343+
}
344+
345+
// Revert to default.
346+
dn.reconfigureProperty(DFS_BLOCKREPORT_INTERVAL_MSEC_KEY,
347+
null);
348+
assertEquals(String.format("%s has wrong value",
349+
DFS_BLOCKREPORT_INTERVAL_MSEC_KEY),
350+
DFS_BLOCKREPORT_INTERVAL_MSEC_DEFAULT,
351+
dn.getDnConf().getBlockReportInterval());
352+
// Verify default.
353+
for (BPOfferService bpos : dn.getAllBpOs()) {
354+
if (bpos != null) {
355+
for (BPServiceActor actor : bpos.getBPServiceActors()) {
356+
assertEquals(String.format("%s has wrong value",
357+
DFS_BLOCKREPORT_INTERVAL_MSEC_KEY),
358+
DFS_BLOCKREPORT_INTERVAL_MSEC_DEFAULT,
359+
actor.getScheduler().getBlockReportIntervalMs());
360+
}
361+
}
362+
}
363+
assertEquals(String.format("expect %s is not configured",
364+
DFS_BLOCKREPORT_INTERVAL_MSEC_KEY), null, dn
365+
.getConf().get(DFS_BLOCKREPORT_INTERVAL_MSEC_KEY));
366+
}
367+
}
296368
}

‎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
@@ -330,7 +330,7 @@ public void testDataNodeGetReconfigurableProperties() throws IOException {
330330
final List<String> outs = Lists.newArrayList();
331331
final List<String> errs = Lists.newArrayList();
332332
getReconfigurableProperties("datanode", address, outs, errs);
333-
assertEquals(3, outs.size());
333+
assertEquals(4, outs.size());
334334
assertEquals(DFSConfigKeys.DFS_DATANODE_DATA_DIR_KEY, outs.get(1));
335335
}
336336

0 commit comments

Comments
 (0)