fix(hstore): JRaft maxEntriesSize configuration parameters do not take effect (#2630)

Co-authored-by: imbajin <jin@apache.org>
This commit is contained in:
YangJiaqi 2024-08-11 13:04:34 +08:00 committed by GitHub
parent df921e915d
commit a038d2341c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 12 additions and 8 deletions

View File

@ -28,7 +28,8 @@ github:
del_branch_on_merge: true
#labels:
enabled_merge_buttons:
merge: false
# TODO: disable it after common merged
merge: true
rebase: true
squash: true
protected_branches:

View File

@ -98,7 +98,7 @@ public class HgStoreEngineOptions {
/**
* The maximum number of entries in AppendEntriesRequest
*/
private final int maxEntriesSize = 256;
private int maxEntriesSize = 256;
/**
* Raft cluster data backlog occurs, rate limiting wait time in milliseconds.
**/

View File

@ -182,6 +182,8 @@ public class AppConfig {
private int maxSegmentFileSize;
@Value("${raft.maxReplicatorInflightMsgs:256}")
private int maxReplicatorInflightMsgs;
@Value("${raft.maxEntriesSize:256}")
private int maxEntriesSize;
}

View File

@ -100,6 +100,7 @@ public class HgStoreNodeService implements RaftTaskHandler {
.isUseRocksDBSegmentLogStorage());
setMaxSegmentFileSize(appConfig.getRaft().getMaxSegmentFileSize());
setMaxReplicatorInflightMsgs(appConfig.getRaft().getMaxReplicatorInflightMsgs());
setMaxEntriesSize(appConfig.getRaft().getMaxEntriesSize());
}});
setFakePdOptions(new FakePdOptions() {{
setStoreList(appConfig.getFakePdConfig().getStoreList());
@ -125,9 +126,9 @@ public class HgStoreNodeService implements RaftTaskHandler {
}
/**
* 添加raft 任务转发数据给raft
* Add raft task, forward data to raft
*
* @return true 表示数据已被提交false表示未提交用于单副本入库减少批次拆分
* @return true means the data has been submitted, false means not submitted, used to reduce batch splitting for single-replica storage
*/
public <Req extends com.google.protobuf.GeneratedMessageV3>
void addRaftTask(byte methodId, String graphName, Integer partitionId, Req req,
@ -140,14 +141,14 @@ public class HgStoreNodeService implements RaftTaskHandler {
}
//
try {
// 序列化
// Serialization
final byte[] buffer = new byte[req.getSerializedSize() + 1];
final CodedOutputStream output = CodedOutputStream.newInstance(buffer);
output.write(methodId);
req.writeTo(output);
output.checkNoSpaceLeft();
output.flush();
// 传送给raft
// Add raft task
storeEngine.addRaftTask(graphName, partitionId,
RaftOperation.create(methodId, buffer, req), closure);
@ -159,7 +160,7 @@ public class HgStoreNodeService implements RaftTaskHandler {
}
/**
* 来自日志的任务一般是follower 或者 日志回滚的任务
* Tasks from logs, generally tasks from followers or log rollbacks
*/
@Override
public boolean invoke(int partId, byte[] request, RaftClosure response) throws
@ -190,7 +191,7 @@ public class HgStoreNodeService implements RaftTaskHandler {
}
/**
* 处理raft传送过来的数据
* Process the data sent by raft
*/
@Override
public boolean invoke(int partId, byte methodId, Object req, RaftClosure response) throws