Skip to content

Commit 4f0e7ed

Browse files
zerolbsonyzerolbsony
committed
Fix issues related to review comments
1 parent 03e7245 commit 4f0e7ed

3 files changed

Lines changed: 141 additions & 8 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderBalanceWithWALBlockIT.java‎

Lines changed: 109 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,9 @@
3737
import org.apache.iotdb.confignode.rpc.thrift.TShowRegionResp;
3838
import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList;
3939
import org.apache.iotdb.consensus.ConsensusFactory;
40+
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils;
4041
import org.apache.iotdb.it.env.EnvFactory;
42+
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
4143
import org.apache.iotdb.it.framework.IoTDBTestRunner;
4244
import org.apache.iotdb.itbase.category.ClusterIT;
4345
import org.apache.iotdb.rpc.TSStatusCode;
@@ -49,12 +51,21 @@
4951
import org.junit.experimental.categories.Category;
5052
import org.junit.runner.RunWith;
5153

54+
import java.io.File;
55+
import java.io.IOException;
56+
import java.nio.file.Files;
57+
import java.nio.file.Path;
58+
import java.sql.Connection;
59+
import java.sql.Statement;
5260
import java.util.ArrayList;
5361
import java.util.Collections;
62+
import java.util.Comparator;
5463
import java.util.HashMap;
5564
import java.util.List;
5665
import java.util.Map;
5766
import java.util.concurrent.TimeUnit;
67+
import java.util.stream.Collectors;
68+
import java.util.stream.Stream;
5869

5970
@RunWith(IoTDBTestRunner.class)
6071
@Category({ClusterIT.class})
@@ -68,9 +79,15 @@ public class IoTDBRegionGroupLeaderBalanceWithWALBlockIT {
6879
private static final int TEST_DATA_NODE_NUM = 3;
6980
private static final int DATABASE_NUM = 3;
7081
private static final int RETRY_NUM = 60;
82+
private static final int TEST_SERIES_PARTITION_SLOT = 0;
83+
private static final long TEST_TIME_PARTITION_SLOT = 0;
84+
private static final int WAL_FILE_SIZE_THRESHOLD_IN_BYTE = 1024;
85+
private static final int WAL_PAYLOAD_REPEAT_COUNT = 128;
7186

7287
private static final String DATABASE = "root.wal_block_db";
7388
private static final String WAL_THROTTLE_THRESHOLD_IN_BYTE = "wal_throttle_threshold_in_byte";
89+
private static final String WAL_FILE_SIZE_THRESHOLD_IN_BYTE_CONFIG =
90+
"wal_file_size_threshold_in_byte";
7491
private static final String WAL_BLOCKED_STATUS = NodeStatus.ReadOnly.getStatus() + "(WALBlocked)";
7592

7693
@Before
@@ -84,6 +101,7 @@ public void setUp() {
84101
.setDataRegionConsensusProtocolClass(TEST_DATA_REGION_CONSENSUS_PROTOCOL_CLASS)
85102
.setSchemaReplicationFactor(TEST_REPLICATION_FACTOR)
86103
.setDataReplicationFactor(TEST_REPLICATION_FACTOR)
104+
.setSeriesSlotNum(1)
87105
.setCheckPeriodWhenInsertBlocked(50)
88106
.setMaxWaitingTimeWhenInsertBlocked(2000);
89107
EnvFactory.getEnv().initClusterEnvironment(1, TEST_DATA_NODE_NUM);
@@ -104,7 +122,8 @@ public void testRegionLeaderBalanceWhenWalLongTermBlocked() throws Exception {
104122
() -> isLeaderDistributionBalanced(client));
105123

106124
TRegionInfo targetLeader = findAnyDataRegionLeader(client);
107-
triggerLongTermWalBlockingOnDataNode(client, targetLeader.getDataNodeId());
125+
long walDiskUsage = generateWalTraffic(client, targetLeader);
126+
triggerLongTermWalBlockingOnDataNode(client, targetLeader.getDataNodeId(), walDiskUsage);
108127

109128
waitUntil(
110129
"target leader DataNode becomes ReadOnly because of long-term WAL blocking",
@@ -124,9 +143,10 @@ private void createDataRegionGroups(SyncConfigNodeIServiceClient client) throws
124143

125144
Map<TSeriesPartitionSlot, TTimeSlotList> seriesSlotMap = new HashMap<>();
126145
seriesSlotMap.put(
127-
new TSeriesPartitionSlot(1),
146+
new TSeriesPartitionSlot(TEST_SERIES_PARTITION_SLOT),
128147
new TTimeSlotList()
129-
.setTimePartitionSlots(Collections.singletonList(new TTimePartitionSlot(100))));
148+
.setTimePartitionSlots(
149+
Collections.singletonList(new TTimePartitionSlot(TEST_TIME_PARTITION_SLOT))));
130150
Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> databaseSlotsMap = new HashMap<>();
131151
databaseSlotsMap.put(DATABASE + i, seriesSlotMap);
132152

@@ -168,13 +188,95 @@ private TRegionInfo findAnyDataRegionLeader(SyncConfigNodeIServiceClient client)
168188
}
169189

170190
private void triggerLongTermWalBlockingOnDataNode(
171-
SyncConfigNodeIServiceClient client, int dataNodeId) throws Exception {
191+
SyncConfigNodeIServiceClient client, int dataNodeId, long walDiskUsage) throws Exception {
192+
Assert.assertTrue("No WAL traffic was generated on target DataNode", walDiskUsage > 0);
193+
194+
Map<String, String> configItems = new HashMap<>();
195+
configItems.put(WAL_THROTTLE_THRESHOLD_IN_BYTE, Long.toString(walDiskUsage));
196+
TSStatus status = client.setConfiguration(new TSetConfigurationReq(configItems, dataNodeId));
197+
Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
198+
}
199+
200+
private long generateWalTraffic(SyncConfigNodeIServiceClient client, TRegionInfo targetLeader)
201+
throws Exception {
202+
int dataNodeId = targetLeader.getDataNodeId();
203+
long originalWalDiskUsage = countWalDiskUsage(dataNodeId);
172204
Map<String, String> configItems = new HashMap<>();
173-
// The throttle threshold used by WALManager is 80% of this value, so 1 makes it 0 and
174-
// deterministically triggers long-term WAL blocking on the target DataNode heartbeat.
175-
configItems.put(WAL_THROTTLE_THRESHOLD_IN_BYTE, "1");
205+
configItems.put(
206+
WAL_FILE_SIZE_THRESHOLD_IN_BYTE_CONFIG, Integer.toString(WAL_FILE_SIZE_THRESHOLD_IN_BYTE));
176207
TSStatus status = client.setConfiguration(new TSetConfigurationReq(configItems, dataNodeId));
177208
Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
209+
210+
DataNodeWrapper dataNodeWrapper =
211+
EnvFactory.getEnv()
212+
.dataNodeIdToWrapper(dataNodeId)
213+
.orElseThrow(() -> new AssertionError("DataNode not found: " + dataNodeId));
214+
215+
try (Connection connection =
216+
EnvFactory.getEnv().getConnectionWithSpecifiedDataNode(dataNodeWrapper);
217+
Statement statement = connection.createStatement()) {
218+
String payload = String.join("", Collections.nCopies(WAL_PAYLOAD_REPEAT_COUNT, "wal_block"));
219+
String device =
220+
targetLeader.getDatabase() + ".d" + targetLeader.getConsensusGroupId().getId();
221+
statement.execute("CREATE TIMESERIES " + device + ".s WITH DATATYPE=TEXT, ENCODING=PLAIN");
222+
for (int i = 0; i < 16; i++) {
223+
statement.execute(
224+
"INSERT INTO "
225+
+ device
226+
+ "(time,s) VALUES("
227+
+ (TEST_TIME_PARTITION_SLOT + i)
228+
+ ", '"
229+
+ payload
230+
+ "')");
231+
}
232+
}
233+
234+
final long[] currentWalDiskUsage = new long[1];
235+
waitUntil(
236+
"target DataNode generates WAL files",
237+
() -> {
238+
currentWalDiskUsage[0] = countWalDiskUsage(dataNodeId);
239+
return currentWalDiskUsage[0] > originalWalDiskUsage;
240+
});
241+
return currentWalDiskUsage[0];
242+
}
243+
244+
private long countWalDiskUsage(int dataNodeId) throws IOException {
245+
DataNodeWrapper dataNodeWrapper =
246+
EnvFactory.getEnv()
247+
.dataNodeIdToWrapper(dataNodeId)
248+
.orElseThrow(() -> new AssertionError("DataNode not found: " + dataNodeId));
249+
Path walDir = new File(dataNodeWrapper.getWalDir()).toPath();
250+
if (!Files.exists(walDir)) {
251+
return 0;
252+
}
253+
try (Stream<Path> paths = Files.walk(walDir)) {
254+
Map<Path, List<Path>> walFilesByDir =
255+
paths
256+
.filter(Files::isRegularFile)
257+
.filter(
258+
path ->
259+
WALFileUtils.walFilenameFilter(
260+
path.getParent().toFile(), path.getFileName().toString()))
261+
.collect(Collectors.groupingBy(Path::getParent));
262+
long walDiskUsage = 0;
263+
for (List<Path> walFiles : walFilesByDir.values()) {
264+
if (walFiles.size() <= 1) {
265+
continue;
266+
}
267+
Path currentWalFile =
268+
Collections.max(
269+
walFiles,
270+
Comparator.comparingLong(
271+
path -> WALFileUtils.parseVersionId(path.getFileName().toString())));
272+
for (Path walFile : walFiles) {
273+
if (!walFile.equals(currentWalFile)) {
274+
walDiskUsage += walFile.toFile().length();
275+
}
276+
}
277+
}
278+
return walDiskUsage;
279+
}
178280
}
179281

180282
private String getNodeStatusWithReason(SyncConfigNodeIServiceClient client, int dataNodeId)

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -283,7 +283,11 @@ public void markWalBufferQueueAvailable() {
283283
}
284284

285285
public long getThrottleThreshold() {
286-
return (long) (config.getThrottleThreshold() * 0.8);
286+
long configuredThrottleThreshold = config.getThrottleThreshold();
287+
if (configuredThrottleThreshold <= 0) {
288+
return Long.MAX_VALUE;
289+
}
290+
return Math.max((long) (configuredThrottleThreshold * 0.8), 1);
287291
}
288292

289293
public long getTotalDiskUsage() {

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManagerTest.java‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,12 +59,14 @@ public class WALManagerTest {
5959
};
6060
private String[] prevWalDirs;
6161
private String prevConsensus;
62+
private long prevThrottleThreshold;
6263
private int prevMaxWaitingTimeWhenInsertBlocked;
6364

6465
@Before
6566
public void setUp() throws Exception {
6667
prevConsensus = config.getDataRegionConsensusProtocolClass();
6768
prevWalDirs = commonConfig.getWalDirs();
69+
prevThrottleThreshold = config.getThrottleThreshold();
6870
prevMaxWaitingTimeWhenInsertBlocked = config.getMaxWaitingTimeWhenInsertBlocked();
6971
config.setDataRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS);
7072
commonConfig.setWalDirs(walDirs);
@@ -79,6 +81,7 @@ public void tearDown() throws Exception {
7981
}
8082
config.setDataRegionConsensusProtocolClass(prevConsensus);
8183
commonConfig.setWalDirs(prevWalDirs);
84+
config.setThrottleThreshold(prevThrottleThreshold);
8285
config.setMaxWaitingTimeWhenInsertBlocked(prevMaxWaitingTimeWhenInsertBlocked);
8386
}
8487

@@ -136,6 +139,30 @@ public void testLongTermWriteBlockedByWalThrottle() {
136139
assertFalse(walManager.isLongTermWriteBlocked());
137140
}
138141

142+
@Test
143+
public void testNonPositiveWalThrottleThresholdIsIgnored() {
144+
WALManager walManager = WALManager.getInstance();
145+
config.setThrottleThreshold(0);
146+
walManager.addTotalDiskUsage(1);
147+
148+
assertEquals(Long.MAX_VALUE, walManager.getThrottleThreshold());
149+
assertFalse(walManager.shouldThrottle());
150+
assertFalse(walManager.isLongTermWriteBlocked());
151+
152+
walManager.clear();
153+
config.setThrottleThreshold(-1);
154+
walManager.addTotalDiskUsage(1);
155+
assertEquals(Long.MAX_VALUE, walManager.getThrottleThreshold());
156+
assertFalse(walManager.shouldThrottle());
157+
158+
walManager.clear();
159+
config.setThrottleThreshold(1);
160+
assertEquals(1, walManager.getThrottleThreshold());
161+
walManager.addTotalDiskUsage(1);
162+
assertTrue(walManager.shouldThrottle());
163+
walManager.clear();
164+
}
165+
139166
@Test
140167
public void testLongTermWriteBlockedByWalBufferQueue() {
141168
WALManager walManager = WALManager.getInstance();

0 commit comments

Comments
 (0)