test/dataflow
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

减少回调

Browse Source
This commit is contained in:
huangsimin 2022-07-26 17:52:58 +08:00
parent 159e25ddc6
commit eea062ce8e
1 changed files with 3 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/main/java/com/yuandian/dataflow/controller/PacketsProcessor.java
View File

@ -67,17 +67,16 @@ public class PacketsProcessor implements RpcProcessor<PacketsProcessor.PacketsRe
work.setTaskQueueSize( work.getTaskQueueSize() - request.packets.size()); work.setTaskQueueSize( work.getTaskQueueSize() - request.packets.size());
work.setUpdateAt(Instant.now()); work.setUpdateAt(Instant.now());
log.debug("workerState taskQueueSize: {} psize: {}", work.getTaskQueueSize(), request.packets.size()); log.debug("workerState taskQueueSize: {} psize: {} state {}", work.getTaskQueueSize(), request.packets.size(), state.getWorkers().size());
var op = new Operate(OperateType.PUT); var op = new Operate(OperateType.PUT);
op.setValue(work); op.setValue(work);
Operate.CallOperate(op, new OperateClosure() { Operate.CallOperate(op, new OperateClosure() {
@Override @Override
public void run(Status status) { public void run(Status status) {
log.info("{}", this.getResponse()); log.info("{}", this.getResponse());
var resp = new RaftResponse(); var resp = new RaftResponse();
resp.setRedirect(StateServerFactory.getStateServer().getNode().getLeaderId()); resp.setSuccess(true);
// resp.setRedirect(StateServerFactory.getStateServer().getNode().getLeaderId());
rpcCtx.sendResponse(resp); rpcCtx.sendResponse(resp);
} }