@Test public void testReassignWithoutExeception() throws IOException { ReceiverAdminClient receiverAdminClient = mockSuccessReceiverAdminClient(); coordinator = new Coordinator(metadataStore, receiverAdminClient); Map<Integer, List<Partition>> preAssignMap = metadataStore.getAssignmentsByCube(cubeName).getAssignments(); Map<Integer, List<Partition>> newAssignMap = new HashMap<>(); newAssignMap.put(1, Lists.newArrayList(p1, p2, p3)); newAssignMap.put(2, Lists.newArrayList(p4, p5)); newAssignMap.put(3, Lists.newArrayList(p6)); CubeAssignment preAssigment = new CubeAssignment(cube.getName(), preAssignMap); CubeAssignment newAssigment = new CubeAssignment(cube.getName(), newAssignMap); coordinator.doReassign(cube, preAssigment, newAssigment); }
void doReassign(CubeInstance cubeInstance, CubeAssignment preAssignments, CubeAssignment newAssignments) { String cubeName = preAssignments.getCubeName(); IStreamingSource streamingSource = StreamingSourceFactory.getStreamingSource(cubeInstance); MapDifference<Integer, List<Partition>> assignDiff = Maps.difference(preAssignments.getAssignments(), newAssignments.getAssignments()); Map<Integer, List<Partition>> sameAssign = assignDiff.entriesInCommon(); Map<Integer, List<Partition>> newAssign = assignDiff.entriesOnlyOnRight(); Map<Integer, List<Partition>> removedAssign = assignDiff.entriesOnlyOnLeft(); List<ISourcePosition> allPositions = Lists.newArrayList(); List<ReplicaSet> successSyncReplicaSet = Lists.newArrayList(); ISourcePosition consumePosition; try { for (Map.Entry<Integer, List<Partition>> assignmentEntry : preAssignments.getAssignments().entrySet()) { Integer replicaSetID = assignmentEntry.getKey(); if (sameAssign.containsKey(replicaSetID)) { logger.info("the assignment is not changed for cube:{}, replicaSet:{}", cubeName, replicaSetID); continue; } ReplicaSet rs = getStreamMetadataStore().getReplicaSet(replicaSetID); ISourcePosition position = syncAndStopConsumersInRs(streamingSource, cubeName, rs); allPositions.add(position); successSyncReplicaSet.add(rs); } consumePosition = streamingSource.getSourcePositionHandler().mergePositions(allPositions, MergeStrategy.KEEP_LARGE); logger.info("the consumer position for cube:{} is:{}", cubeName, consumePosition); } catch (Exception e) { logger.error("fail to sync assign replicaSet for cube:" + cubeName, e); Set<Integer> needRollback = successSyncReplicaSet.stream().map(ReplicaSet::getReplicaSetID) .collect(Collectors.toSet()); for (ReplicaSet rs : successSyncReplicaSet) { StartConsumersRequest request = new StartConsumersRequest(); request.setCube(cubeName); try { startConsumersInReplicaSet(rs, request); needRollback.remove(rs.getReplicaSetID()); } catch (IOException e1) { logger.error("fail to start consumers for cube:" + cubeName + " replicaSet:" + rs.getReplicaSetID(), e1); } } if (needRollback.isEmpty()) { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_SUCCESS, TransactionStep.STOP_AND_SNYC, "", e); } else { StringBuilder str = new StringBuilder(); try { str.append("Fail restart:").append(JsonUtil.writeValueAsString(needRollback)); } catch (JsonProcessingException jpe) { logger.error("", jpe); } throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_FAILED, TransactionStep.STOP_AND_SNYC, str.toString(), e); } } List<ReplicaSet> successAssigned = Lists.newArrayList(); List<ReplicaSet> successStarted = Lists.newArrayList(); Set<Node> failedConvertToImmutableNodes = new HashSet<>(); try { for (Map.Entry<Integer, List<Partition>> cubeAssignmentEntry : newAssignments.getAssignments().entrySet()) { Integer replicaSetID = cubeAssignmentEntry.getKey(); if (sameAssign.containsKey(replicaSetID)) { continue; } ReplicaSet rs = getStreamMetadataStore().getReplicaSet(replicaSetID); logger.info("assign cube:{} to replicaSet:{}", cubeName, replicaSetID); assignCubeToReplicaSet(rs, cubeName, cubeAssignmentEntry.getValue(), false, true); successAssigned.add(rs); } for (Map.Entry<Integer, List<Partition>> cubeAssignmentEntry : newAssignments.getAssignments().entrySet()) { Integer replicaSetID = cubeAssignmentEntry.getKey(); if (sameAssign.containsKey(replicaSetID)) { continue; } ConsumerStartProtocol consumerStartProtocol = new ConsumerStartProtocol( streamingSource.getSourcePositionHandler().serializePosition(consumePosition.advance())); ReplicaSet rs = getStreamMetadataStore().getReplicaSet(replicaSetID); StartConsumersRequest startRequest = new StartConsumersRequest(); startRequest.setCube(cubeName); startRequest.setStartProtocol(consumerStartProtocol); logger.info("start consumers for cube:{}, replicaSet:{}, startRequest:{}", cubeName, replicaSetID, startRequest); startConsumersInReplicaSet(rs, startRequest); successStarted.add(rs); } for (Map.Entry<Integer, List<Partition>> removeAssignmentEntry : removedAssign.entrySet()) { Integer replicaSetID = removeAssignmentEntry.getKey(); logger.info("make cube immutable for cube:{}, replicaSet{}", cubeName, replicaSetID); ReplicaSet rs = getStreamMetadataStore().getReplicaSet(replicaSetID); List<Node> failedNodes = makeCubeImmutableInReplicaSet(rs, cubeName); failedConvertToImmutableNodes.addAll(failedNodes); } if (!failedConvertToImmutableNodes.isEmpty()) { throw new IOException(); } logger.info("finish cube reBalance, cube:{}", cubeName); } catch (IOException e) { logger.error("fail to start consumers for cube:" + cubeName, e); Set<Integer> rollbackStarted = successStarted.stream().map(ReplicaSet::getReplicaSetID) .collect(Collectors.toSet()); for (ReplicaSet rs : successStarted) { try { StopConsumersRequest stopRequest = new StopConsumersRequest(); stopRequest.setCube(cubeName); if (newAssign.containsKey(rs.getReplicaSetID())) { stopRequest.setRemoveData(true); } stopConsumersInReplicaSet(rs, stopRequest); rollbackStarted.remove(rs.getReplicaSetID()); } catch (IOException e1) { logger.error("fail to stop consumers for cube:" + cubeName + " replicaSet:" + rs.getReplicaSetID(), e1); } } Set<Integer> rollbackAssigned = successAssigned.stream().map(ReplicaSet::getReplicaSetID) .collect(Collectors.toSet()); for (ReplicaSet rs : successAssigned) { try { List<Partition> partitions = preAssignments.getPartitionsByReplicaSetID(rs.getReplicaSetID()); assignCubeToReplicaSet(rs, cubeName, partitions, true, true); rollbackAssigned.remove(rs.getReplicaSetID()); } catch (IOException e1) { logger.error("fail to start consumers for cube:" + cubeName + " replicaSet:" + rs.getReplicaSetID(), e1); } } Set<Node> failedReceiver = new HashSet<>(failedConvertToImmutableNodes); for (Node node : failedConvertToImmutableNodes) { try { makeCubeImmutableForReceiver(node, cubeName); failedReceiver.remove(node); } catch (IOException ioe) { logger.error("fail to make cube immutable for cube:" + cubeName + " to " + node, ioe); } } StringBuilder str = new StringBuilder(); try { str.append("FailStarted:").append(JsonUtil.writeValueAsString(rollbackStarted)).append(";"); str.append("FailAssigned:").append(JsonUtil.writeValueAsString(rollbackAssigned)).append(";"); str.append("FailRemotedPresisted:").append(JsonUtil.writeValueAsString(failedReceiver)); } catch (JsonProcessingException jpe) { logger.error("", jpe); } String failedInfo = str.toString(); if (!rollbackStarted.isEmpty()) { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_FAILED, TransactionStep.START_NEW, failedInfo, e); } else if (!rollbackAssigned.isEmpty()) { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_FAILED, TransactionStep.ASSIGN_NEW, failedInfo, e); } else if (!failedReceiver.isEmpty()) { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_FAILED, TransactionStep.MAKE_IMMUTABLE, failedInfo, e); } else { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_SUCCESS, TransactionStep.ASSIGN_NEW, failedInfo, e); } } }
Coordinator implements CoordinatorClient { void doReassign(CubeInstance cubeInstance, CubeAssignment preAssignments, CubeAssignment newAssignments) { String cubeName = preAssignments.getCubeName(); IStreamingSource streamingSource = StreamingSourceFactory.getStreamingSource(cubeInstance); MapDifference<Integer, List<Partition>> assignDiff = Maps.difference(preAssignments.getAssignments(), newAssignments.getAssignments()); Map<Integer, List<Partition>> sameAssign = assignDiff.entriesInCommon(); Map<Integer, List<Partition>> newAssign = assignDiff.entriesOnlyOnRight(); Map<Integer, List<Partition>> removedAssign = assignDiff.entriesOnlyOnLeft(); List<ISourcePosition> allPositions = Lists.newArrayList(); List<ReplicaSet> successSyncReplicaSet = Lists.newArrayList(); ISourcePosition consumePosition; try { for (Map.Entry<Integer, List<Partition>> assignmentEntry : preAssignments.getAssignments().entrySet()) { Integer replicaSetID = assignmentEntry.getKey(); if (sameAssign.containsKey(replicaSetID)) { logger.info("the assignment is not changed for cube:{}, replicaSet:{}", cubeName, replicaSetID); continue; } ReplicaSet rs = getStreamMetadataStore().getReplicaSet(replicaSetID); ISourcePosition position = syncAndStopConsumersInRs(streamingSource, cubeName, rs); allPositions.add(position); successSyncReplicaSet.add(rs); } consumePosition = streamingSource.getSourcePositionHandler().mergePositions(allPositions, MergeStrategy.KEEP_LARGE); logger.info("the consumer position for cube:{} is:{}", cubeName, consumePosition); } catch (Exception e) { logger.error("fail to sync assign replicaSet for cube:" + cubeName, e); Set<Integer> needRollback = successSyncReplicaSet.stream().map(ReplicaSet::getReplicaSetID) .collect(Collectors.toSet()); for (ReplicaSet rs : successSyncReplicaSet) { StartConsumersRequest request = new StartConsumersRequest(); request.setCube(cubeName); try { startConsumersInReplicaSet(rs, request); needRollback.remove(rs.getReplicaSetID()); } catch (IOException e1) { logger.error("fail to start consumers for cube:" + cubeName + " replicaSet:" + rs.getReplicaSetID(), e1); } } if (needRollback.isEmpty()) { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_SUCCESS, TransactionStep.STOP_AND_SNYC, "", e); } else { StringBuilder str = new StringBuilder(); try { str.append("Fail restart:").append(JsonUtil.writeValueAsString(needRollback)); } catch (JsonProcessingException jpe) { logger.error("", jpe); } throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_FAILED, TransactionStep.STOP_AND_SNYC, str.toString(), e); } } List<ReplicaSet> successAssigned = Lists.newArrayList(); List<ReplicaSet> successStarted = Lists.newArrayList(); Set<Node> failedConvertToImmutableNodes = new HashSet<>(); try { for (Map.Entry<Integer, List<Partition>> cubeAssignmentEntry : newAssignments.getAssignments().entrySet()) { Integer replicaSetID = cubeAssignmentEntry.getKey(); if (sameAssign.containsKey(replicaSetID)) { continue; } ReplicaSet rs = getStreamMetadataStore().getReplicaSet(replicaSetID); logger.info("assign cube:{} to replicaSet:{}", cubeName, replicaSetID); assignCubeToReplicaSet(rs, cubeName, cubeAssignmentEntry.getValue(), false, true); successAssigned.add(rs); } for (Map.Entry<Integer, List<Partition>> cubeAssignmentEntry : newAssignments.getAssignments().entrySet()) { Integer replicaSetID = cubeAssignmentEntry.getKey(); if (sameAssign.containsKey(replicaSetID)) { continue; } ConsumerStartProtocol consumerStartProtocol = new ConsumerStartProtocol( streamingSource.getSourcePositionHandler().serializePosition(consumePosition.advance())); ReplicaSet rs = getStreamMetadataStore().getReplicaSet(replicaSetID); StartConsumersRequest startRequest = new StartConsumersRequest(); startRequest.setCube(cubeName); startRequest.setStartProtocol(consumerStartProtocol); logger.info("start consumers for cube:{}, replicaSet:{}, startRequest:{}", cubeName, replicaSetID, startRequest); startConsumersInReplicaSet(rs, startRequest); successStarted.add(rs); } for (Map.Entry<Integer, List<Partition>> removeAssignmentEntry : removedAssign.entrySet()) { Integer replicaSetID = removeAssignmentEntry.getKey(); logger.info("make cube immutable for cube:{}, replicaSet{}", cubeName, replicaSetID); ReplicaSet rs = getStreamMetadataStore().getReplicaSet(replicaSetID); List<Node> failedNodes = makeCubeImmutableInReplicaSet(rs, cubeName); failedConvertToImmutableNodes.addAll(failedNodes); } if (!failedConvertToImmutableNodes.isEmpty()) { throw new IOException(); } logger.info("finish cube reBalance, cube:{}", cubeName); } catch (IOException e) { logger.error("fail to start consumers for cube:" + cubeName, e); Set<Integer> rollbackStarted = successStarted.stream().map(ReplicaSet::getReplicaSetID) .collect(Collectors.toSet()); for (ReplicaSet rs : successStarted) { try { StopConsumersRequest stopRequest = new StopConsumersRequest(); stopRequest.setCube(cubeName); if (newAssign.containsKey(rs.getReplicaSetID())) { stopRequest.setRemoveData(true); } stopConsumersInReplicaSet(rs, stopRequest); rollbackStarted.remove(rs.getReplicaSetID()); } catch (IOException e1) { logger.error("fail to stop consumers for cube:" + cubeName + " replicaSet:" + rs.getReplicaSetID(), e1); } } Set<Integer> rollbackAssigned = successAssigned.stream().map(ReplicaSet::getReplicaSetID) .collect(Collectors.toSet()); for (ReplicaSet rs : successAssigned) { try { List<Partition> partitions = preAssignments.getPartitionsByReplicaSetID(rs.getReplicaSetID()); assignCubeToReplicaSet(rs, cubeName, partitions, true, true); rollbackAssigned.remove(rs.getReplicaSetID()); } catch (IOException e1) { logger.error("fail to start consumers for cube:" + cubeName + " replicaSet:" + rs.getReplicaSetID(), e1); } } Set<Node> failedReceiver = new HashSet<>(failedConvertToImmutableNodes); for (Node node : failedConvertToImmutableNodes) { try { makeCubeImmutableForReceiver(node, cubeName); failedReceiver.remove(node); } catch (IOException ioe) { logger.error("fail to make cube immutable for cube:" + cubeName + " to " + node, ioe); } } StringBuilder str = new StringBuilder(); try { str.append("FailStarted:").append(JsonUtil.writeValueAsString(rollbackStarted)).append(";"); str.append("FailAssigned:").append(JsonUtil.writeValueAsString(rollbackAssigned)).append(";"); str.append("FailRemotedPresisted:").append(JsonUtil.writeValueAsString(failedReceiver)); } catch (JsonProcessingException jpe) { logger.error("", jpe); } String failedInfo = str.toString(); if (!rollbackStarted.isEmpty()) { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_FAILED, TransactionStep.START_NEW, failedInfo, e); } else if (!rollbackAssigned.isEmpty()) { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_FAILED, TransactionStep.ASSIGN_NEW, failedInfo, e); } else if (!failedReceiver.isEmpty()) { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_FAILED, TransactionStep.MAKE_IMMUTABLE, failedInfo, e); } else { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_SUCCESS, TransactionStep.ASSIGN_NEW, failedInfo, e); } } } }
Coordinator implements CoordinatorClient { void doReassign(CubeInstance cubeInstance, CubeAssignment preAssignments, CubeAssignment newAssignments) { String cubeName = preAssignments.getCubeName(); IStreamingSource streamingSource = StreamingSourceFactory.getStreamingSource(cubeInstance); MapDifference<Integer, List<Partition>> assignDiff = Maps.difference(preAssignments.getAssignments(), newAssignments.getAssignments()); Map<Integer, List<Partition>> sameAssign = assignDiff.entriesInCommon(); Map<Integer, List<Partition>> newAssign = assignDiff.entriesOnlyOnRight(); Map<Integer, List<Partition>> removedAssign = assignDiff.entriesOnlyOnLeft(); List<ISourcePosition> allPositions = Lists.newArrayList(); List<ReplicaSet> successSyncReplicaSet = Lists.newArrayList(); ISourcePosition consumePosition; try { for (Map.Entry<Integer, List<Partition>> assignmentEntry : preAssignments.getAssignments().entrySet()) { Integer replicaSetID = assignmentEntry.getKey(); if (sameAssign.containsKey(replicaSetID)) { logger.info("the assignment is not changed for cube:{}, replicaSet:{}", cubeName, replicaSetID); continue; } ReplicaSet rs = getStreamMetadataStore().getReplicaSet(replicaSetID); ISourcePosition position = syncAndStopConsumersInRs(streamingSource, cubeName, rs); allPositions.add(position); successSyncReplicaSet.add(rs); } consumePosition = streamingSource.getSourcePositionHandler().mergePositions(allPositions, MergeStrategy.KEEP_LARGE); logger.info("the consumer position for cube:{} is:{}", cubeName, consumePosition); } catch (Exception e) { logger.error("fail to sync assign replicaSet for cube:" + cubeName, e); Set<Integer> needRollback = successSyncReplicaSet.stream().map(ReplicaSet::getReplicaSetID) .collect(Collectors.toSet()); for (ReplicaSet rs : successSyncReplicaSet) { StartConsumersRequest request = new StartConsumersRequest(); request.setCube(cubeName); try { startConsumersInReplicaSet(rs, request); needRollback.remove(rs.getReplicaSetID()); } catch (IOException e1) { logger.error("fail to start consumers for cube:" + cubeName + " replicaSet:" + rs.getReplicaSetID(), e1); } } if (needRollback.isEmpty()) { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_SUCCESS, TransactionStep.STOP_AND_SNYC, "", e); } else { StringBuilder str = new StringBuilder(); try { str.append("Fail restart:").append(JsonUtil.writeValueAsString(needRollback)); } catch (JsonProcessingException jpe) { logger.error("", jpe); } throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_FAILED, TransactionStep.STOP_AND_SNYC, str.toString(), e); } } List<ReplicaSet> successAssigned = Lists.newArrayList(); List<ReplicaSet> successStarted = Lists.newArrayList(); Set<Node> failedConvertToImmutableNodes = new HashSet<>(); try { for (Map.Entry<Integer, List<Partition>> cubeAssignmentEntry : newAssignments.getAssignments().entrySet()) { Integer replicaSetID = cubeAssignmentEntry.getKey(); if (sameAssign.containsKey(replicaSetID)) { continue; } ReplicaSet rs = getStreamMetadataStore().getReplicaSet(replicaSetID); logger.info("assign cube:{} to replicaSet:{}", cubeName, replicaSetID); assignCubeToReplicaSet(rs, cubeName, cubeAssignmentEntry.getValue(), false, true); successAssigned.add(rs); } for (Map.Entry<Integer, List<Partition>> cubeAssignmentEntry : newAssignments.getAssignments().entrySet()) { Integer replicaSetID = cubeAssignmentEntry.getKey(); if (sameAssign.containsKey(replicaSetID)) { continue; } ConsumerStartProtocol consumerStartProtocol = new ConsumerStartProtocol( streamingSource.getSourcePositionHandler().serializePosition(consumePosition.advance())); ReplicaSet rs = getStreamMetadataStore().getReplicaSet(replicaSetID); StartConsumersRequest startRequest = new StartConsumersRequest(); startRequest.setCube(cubeName); startRequest.setStartProtocol(consumerStartProtocol); logger.info("start consumers for cube:{}, replicaSet:{}, startRequest:{}", cubeName, replicaSetID, startRequest); startConsumersInReplicaSet(rs, startRequest); successStarted.add(rs); } for (Map.Entry<Integer, List<Partition>> removeAssignmentEntry : removedAssign.entrySet()) { Integer replicaSetID = removeAssignmentEntry.getKey(); logger.info("make cube immutable for cube:{}, replicaSet{}", cubeName, replicaSetID); ReplicaSet rs = getStreamMetadataStore().getReplicaSet(replicaSetID); List<Node> failedNodes = makeCubeImmutableInReplicaSet(rs, cubeName); failedConvertToImmutableNodes.addAll(failedNodes); } if (!failedConvertToImmutableNodes.isEmpty()) { throw new IOException(); } logger.info("finish cube reBalance, cube:{}", cubeName); } catch (IOException e) { logger.error("fail to start consumers for cube:" + cubeName, e); Set<Integer> rollbackStarted = successStarted.stream().map(ReplicaSet::getReplicaSetID) .collect(Collectors.toSet()); for (ReplicaSet rs : successStarted) { try { StopConsumersRequest stopRequest = new StopConsumersRequest(); stopRequest.setCube(cubeName); if (newAssign.containsKey(rs.getReplicaSetID())) { stopRequest.setRemoveData(true); } stopConsumersInReplicaSet(rs, stopRequest); rollbackStarted.remove(rs.getReplicaSetID()); } catch (IOException e1) { logger.error("fail to stop consumers for cube:" + cubeName + " replicaSet:" + rs.getReplicaSetID(), e1); } } Set<Integer> rollbackAssigned = successAssigned.stream().map(ReplicaSet::getReplicaSetID) .collect(Collectors.toSet()); for (ReplicaSet rs : successAssigned) { try { List<Partition> partitions = preAssignments.getPartitionsByReplicaSetID(rs.getReplicaSetID()); assignCubeToReplicaSet(rs, cubeName, partitions, true, true); rollbackAssigned.remove(rs.getReplicaSetID()); } catch (IOException e1) { logger.error("fail to start consumers for cube:" + cubeName + " replicaSet:" + rs.getReplicaSetID(), e1); } } Set<Node> failedReceiver = new HashSet<>(failedConvertToImmutableNodes); for (Node node : failedConvertToImmutableNodes) { try { makeCubeImmutableForReceiver(node, cubeName); failedReceiver.remove(node); } catch (IOException ioe) { logger.error("fail to make cube immutable for cube:" + cubeName + " to " + node, ioe); } } StringBuilder str = new StringBuilder(); try { str.append("FailStarted:").append(JsonUtil.writeValueAsString(rollbackStarted)).append(";"); str.append("FailAssigned:").append(JsonUtil.writeValueAsString(rollbackAssigned)).append(";"); str.append("FailRemotedPresisted:").append(JsonUtil.writeValueAsString(failedReceiver)); } catch (JsonProcessingException jpe) { logger.error("", jpe); } String failedInfo = str.toString(); if (!rollbackStarted.isEmpty()) { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_FAILED, TransactionStep.START_NEW, failedInfo, e); } else if (!rollbackAssigned.isEmpty()) { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_FAILED, TransactionStep.ASSIGN_NEW, failedInfo, e); } else if (!failedReceiver.isEmpty()) { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_FAILED, TransactionStep.MAKE_IMMUTABLE, failedInfo, e); } else { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_SUCCESS, TransactionStep.ASSIGN_NEW, failedInfo, e); } } } private  Coordinator(); @VisibleForTesting  Coordinator(StreamMetadataStore metadataStore, ReceiverAdminClient receiverClient); }
Coordinator implements CoordinatorClient { void doReassign(CubeInstance cubeInstance, CubeAssignment preAssignments, CubeAssignment newAssignments) { String cubeName = preAssignments.getCubeName(); IStreamingSource streamingSource = StreamingSourceFactory.getStreamingSource(cubeInstance); MapDifference<Integer, List<Partition>> assignDiff = Maps.difference(preAssignments.getAssignments(), newAssignments.getAssignments()); Map<Integer, List<Partition>> sameAssign = assignDiff.entriesInCommon(); Map<Integer, List<Partition>> newAssign = assignDiff.entriesOnlyOnRight(); Map<Integer, List<Partition>> removedAssign = assignDiff.entriesOnlyOnLeft(); List<ISourcePosition> allPositions = Lists.newArrayList(); List<ReplicaSet> successSyncReplicaSet = Lists.newArrayList(); ISourcePosition consumePosition; try { for (Map.Entry<Integer, List<Partition>> assignmentEntry : preAssignments.getAssignments().entrySet()) { Integer replicaSetID = assignmentEntry.getKey(); if (sameAssign.containsKey(replicaSetID)) { logger.info("the assignment is not changed for cube:{}, replicaSet:{}", cubeName, replicaSetID); continue; } ReplicaSet rs = getStreamMetadataStore().getReplicaSet(replicaSetID); ISourcePosition position = syncAndStopConsumersInRs(streamingSource, cubeName, rs); allPositions.add(position); successSyncReplicaSet.add(rs); } consumePosition = streamingSource.getSourcePositionHandler().mergePositions(allPositions, MergeStrategy.KEEP_LARGE); logger.info("the consumer position for cube:{} is:{}", cubeName, consumePosition); } catch (Exception e) { logger.error("fail to sync assign replicaSet for cube:" + cubeName, e); Set<Integer> needRollback = successSyncReplicaSet.stream().map(ReplicaSet::getReplicaSetID) .collect(Collectors.toSet()); for (ReplicaSet rs : successSyncReplicaSet) { StartConsumersRequest request = new StartConsumersRequest(); request.setCube(cubeName); try { startConsumersInReplicaSet(rs, request); needRollback.remove(rs.getReplicaSetID()); } catch (IOException e1) { logger.error("fail to start consumers for cube:" + cubeName + " replicaSet:" + rs.getReplicaSetID(), e1); } } if (needRollback.isEmpty()) { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_SUCCESS, TransactionStep.STOP_AND_SNYC, "", e); } else { StringBuilder str = new StringBuilder(); try { str.append("Fail restart:").append(JsonUtil.writeValueAsString(needRollback)); } catch (JsonProcessingException jpe) { logger.error("", jpe); } throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_FAILED, TransactionStep.STOP_AND_SNYC, str.toString(), e); } } List<ReplicaSet> successAssigned = Lists.newArrayList(); List<ReplicaSet> successStarted = Lists.newArrayList(); Set<Node> failedConvertToImmutableNodes = new HashSet<>(); try { for (Map.Entry<Integer, List<Partition>> cubeAssignmentEntry : newAssignments.getAssignments().entrySet()) { Integer replicaSetID = cubeAssignmentEntry.getKey(); if (sameAssign.containsKey(replicaSetID)) { continue; } ReplicaSet rs = getStreamMetadataStore().getReplicaSet(replicaSetID); logger.info("assign cube:{} to replicaSet:{}", cubeName, replicaSetID); assignCubeToReplicaSet(rs, cubeName, cubeAssignmentEntry.getValue(), false, true); successAssigned.add(rs); } for (Map.Entry<Integer, List<Partition>> cubeAssignmentEntry : newAssignments.getAssignments().entrySet()) { Integer replicaSetID = cubeAssignmentEntry.getKey(); if (sameAssign.containsKey(replicaSetID)) { continue; } ConsumerStartProtocol consumerStartProtocol = new ConsumerStartProtocol( streamingSource.getSourcePositionHandler().serializePosition(consumePosition.advance())); ReplicaSet rs = getStreamMetadataStore().getReplicaSet(replicaSetID); StartConsumersRequest startRequest = new StartConsumersRequest(); startRequest.setCube(cubeName); startRequest.setStartProtocol(consumerStartProtocol); logger.info("start consumers for cube:{}, replicaSet:{}, startRequest:{}", cubeName, replicaSetID, startRequest); startConsumersInReplicaSet(rs, startRequest); successStarted.add(rs); } for (Map.Entry<Integer, List<Partition>> removeAssignmentEntry : removedAssign.entrySet()) { Integer replicaSetID = removeAssignmentEntry.getKey(); logger.info("make cube immutable for cube:{}, replicaSet{}", cubeName, replicaSetID); ReplicaSet rs = getStreamMetadataStore().getReplicaSet(replicaSetID); List<Node> failedNodes = makeCubeImmutableInReplicaSet(rs, cubeName); failedConvertToImmutableNodes.addAll(failedNodes); } if (!failedConvertToImmutableNodes.isEmpty()) { throw new IOException(); } logger.info("finish cube reBalance, cube:{}", cubeName); } catch (IOException e) { logger.error("fail to start consumers for cube:" + cubeName, e); Set<Integer> rollbackStarted = successStarted.stream().map(ReplicaSet::getReplicaSetID) .collect(Collectors.toSet()); for (ReplicaSet rs : successStarted) { try { StopConsumersRequest stopRequest = new StopConsumersRequest(); stopRequest.setCube(cubeName); if (newAssign.containsKey(rs.getReplicaSetID())) { stopRequest.setRemoveData(true); } stopConsumersInReplicaSet(rs, stopRequest); rollbackStarted.remove(rs.getReplicaSetID()); } catch (IOException e1) { logger.error("fail to stop consumers for cube:" + cubeName + " replicaSet:" + rs.getReplicaSetID(), e1); } } Set<Integer> rollbackAssigned = successAssigned.stream().map(ReplicaSet::getReplicaSetID) .collect(Collectors.toSet()); for (ReplicaSet rs : successAssigned) { try { List<Partition> partitions = preAssignments.getPartitionsByReplicaSetID(rs.getReplicaSetID()); assignCubeToReplicaSet(rs, cubeName, partitions, true, true); rollbackAssigned.remove(rs.getReplicaSetID()); } catch (IOException e1) { logger.error("fail to start consumers for cube:" + cubeName + " replicaSet:" + rs.getReplicaSetID(), e1); } } Set<Node> failedReceiver = new HashSet<>(failedConvertToImmutableNodes); for (Node node : failedConvertToImmutableNodes) { try { makeCubeImmutableForReceiver(node, cubeName); failedReceiver.remove(node); } catch (IOException ioe) { logger.error("fail to make cube immutable for cube:" + cubeName + " to " + node, ioe); } } StringBuilder str = new StringBuilder(); try { str.append("FailStarted:").append(JsonUtil.writeValueAsString(rollbackStarted)).append(";"); str.append("FailAssigned:").append(JsonUtil.writeValueAsString(rollbackAssigned)).append(";"); str.append("FailRemotedPresisted:").append(JsonUtil.writeValueAsString(failedReceiver)); } catch (JsonProcessingException jpe) { logger.error("", jpe); } String failedInfo = str.toString(); if (!rollbackStarted.isEmpty()) { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_FAILED, TransactionStep.START_NEW, failedInfo, e); } else if (!rollbackAssigned.isEmpty()) { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_FAILED, TransactionStep.ASSIGN_NEW, failedInfo, e); } else if (!failedReceiver.isEmpty()) { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_FAILED, TransactionStep.MAKE_IMMUTABLE, failedInfo, e); } else { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_SUCCESS, TransactionStep.ASSIGN_NEW, failedInfo, e); } } } private  Coordinator(); @VisibleForTesting  Coordinator(StreamMetadataStore metadataStore, ReceiverAdminClient receiverClient); static Coordinator getInstance(); void start(); @VisibleForTesting void setReceiverAdminClient(ReceiverAdminClient receiverAdminClient); @VisibleForTesting void setToLeader(); @Override synchronized void assignCube(String cubeName); @Override void unAssignCube(String cubeName); @Override synchronized void reAssignCube(String cubeName, CubeAssignment assignments); @Override void segmentRemoteStoreComplete(Node receiver, String cubeName, Pair<Long, Long> segment); @Override Map<Integer, Map<String, List<Partition>>> reBalanceRecommend(); @Override synchronized void reBalance(Map<Integer, Map<String, List<Partition>>> newAssignmentsPlan); @Override void pauseConsumers(String cubeName); @Override void resumeConsumers(String cubeName); @Override void replicaSetLeaderChange(int replicaSetID, Node newLeader); StreamingCubeInfo getStreamCubeInfo(String cubeName); Map<Integer, Map<String, List<Partition>>> reBalancePlan(List<StreamingCubeInfo> allCubes,
            List<ReplicaSet> allReplicaSets); void removeCubeHDFSFiles(String cubeName); synchronized void createReplicaSet(ReplicaSet rs); synchronized void removeReplicaSet(int rsID); synchronized void addNodeToReplicaSet(Integer replicaSetID, String nodeID); synchronized void removeNodeFromReplicaSet(Integer replicaSetID, String nodeID); void stopConsumers(String cubeName); void assignCubeToReplicaSet(ReplicaSet rs, String cubeName, List<Partition> partitions,
            boolean startConsumer, boolean mustAllSucceed); ConsumerStatsResponse resumeConsumersForReceiver(final Node receiver, final ResumeConsumerRequest request); void startConsumersInReplicaSet(ReplicaSet rs, final StartConsumersRequest request); List<Node> makeCubeImmutableInReplicaSet(ReplicaSet rs, String cubeName); List<ConsumerStatsResponse> stopConsumersInReplicaSet(ReplicaSet rs, final StopConsumersRequest request); List<ConsumerStatsResponse> pauseConsumersInReplicaSet(ReplicaSet rs, final PauseConsumersRequest request); List<ConsumerStatsResponse> resumeConsumersInReplicaSet(ReplicaSet rs, final ResumeConsumerRequest request); StreamMetadataStore getStreamMetadataStore(); ExecutableManager getExecutableManager(); CubeManager getCubeManager(); }
Coordinator implements CoordinatorClient { void doReassign(CubeInstance cubeInstance, CubeAssignment preAssignments, CubeAssignment newAssignments) { String cubeName = preAssignments.getCubeName(); IStreamingSource streamingSource = StreamingSourceFactory.getStreamingSource(cubeInstance); MapDifference<Integer, List<Partition>> assignDiff = Maps.difference(preAssignments.getAssignments(), newAssignments.getAssignments()); Map<Integer, List<Partition>> sameAssign = assignDiff.entriesInCommon(); Map<Integer, List<Partition>> newAssign = assignDiff.entriesOnlyOnRight(); Map<Integer, List<Partition>> removedAssign = assignDiff.entriesOnlyOnLeft(); List<ISourcePosition> allPositions = Lists.newArrayList(); List<ReplicaSet> successSyncReplicaSet = Lists.newArrayList(); ISourcePosition consumePosition; try { for (Map.Entry<Integer, List<Partition>> assignmentEntry : preAssignments.getAssignments().entrySet()) { Integer replicaSetID = assignmentEntry.getKey(); if (sameAssign.containsKey(replicaSetID)) { logger.info("the assignment is not changed for cube:{}, replicaSet:{}", cubeName, replicaSetID); continue; } ReplicaSet rs = getStreamMetadataStore().getReplicaSet(replicaSetID); ISourcePosition position = syncAndStopConsumersInRs(streamingSource, cubeName, rs); allPositions.add(position); successSyncReplicaSet.add(rs); } consumePosition = streamingSource.getSourcePositionHandler().mergePositions(allPositions, MergeStrategy.KEEP_LARGE); logger.info("the consumer position for cube:{} is:{}", cubeName, consumePosition); } catch (Exception e) { logger.error("fail to sync assign replicaSet for cube:" + cubeName, e); Set<Integer> needRollback = successSyncReplicaSet.stream().map(ReplicaSet::getReplicaSetID) .collect(Collectors.toSet()); for (ReplicaSet rs : successSyncReplicaSet) { StartConsumersRequest request = new StartConsumersRequest(); request.setCube(cubeName); try { startConsumersInReplicaSet(rs, request); needRollback.remove(rs.getReplicaSetID()); } catch (IOException e1) { logger.error("fail to start consumers for cube:" + cubeName + " replicaSet:" + rs.getReplicaSetID(), e1); } } if (needRollback.isEmpty()) { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_SUCCESS, TransactionStep.STOP_AND_SNYC, "", e); } else { StringBuilder str = new StringBuilder(); try { str.append("Fail restart:").append(JsonUtil.writeValueAsString(needRollback)); } catch (JsonProcessingException jpe) { logger.error("", jpe); } throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_FAILED, TransactionStep.STOP_AND_SNYC, str.toString(), e); } } List<ReplicaSet> successAssigned = Lists.newArrayList(); List<ReplicaSet> successStarted = Lists.newArrayList(); Set<Node> failedConvertToImmutableNodes = new HashSet<>(); try { for (Map.Entry<Integer, List<Partition>> cubeAssignmentEntry : newAssignments.getAssignments().entrySet()) { Integer replicaSetID = cubeAssignmentEntry.getKey(); if (sameAssign.containsKey(replicaSetID)) { continue; } ReplicaSet rs = getStreamMetadataStore().getReplicaSet(replicaSetID); logger.info("assign cube:{} to replicaSet:{}", cubeName, replicaSetID); assignCubeToReplicaSet(rs, cubeName, cubeAssignmentEntry.getValue(), false, true); successAssigned.add(rs); } for (Map.Entry<Integer, List<Partition>> cubeAssignmentEntry : newAssignments.getAssignments().entrySet()) { Integer replicaSetID = cubeAssignmentEntry.getKey(); if (sameAssign.containsKey(replicaSetID)) { continue; } ConsumerStartProtocol consumerStartProtocol = new ConsumerStartProtocol( streamingSource.getSourcePositionHandler().serializePosition(consumePosition.advance())); ReplicaSet rs = getStreamMetadataStore().getReplicaSet(replicaSetID); StartConsumersRequest startRequest = new StartConsumersRequest(); startRequest.setCube(cubeName); startRequest.setStartProtocol(consumerStartProtocol); logger.info("start consumers for cube:{}, replicaSet:{}, startRequest:{}", cubeName, replicaSetID, startRequest); startConsumersInReplicaSet(rs, startRequest); successStarted.add(rs); } for (Map.Entry<Integer, List<Partition>> removeAssignmentEntry : removedAssign.entrySet()) { Integer replicaSetID = removeAssignmentEntry.getKey(); logger.info("make cube immutable for cube:{}, replicaSet{}", cubeName, replicaSetID); ReplicaSet rs = getStreamMetadataStore().getReplicaSet(replicaSetID); List<Node> failedNodes = makeCubeImmutableInReplicaSet(rs, cubeName); failedConvertToImmutableNodes.addAll(failedNodes); } if (!failedConvertToImmutableNodes.isEmpty()) { throw new IOException(); } logger.info("finish cube reBalance, cube:{}", cubeName); } catch (IOException e) { logger.error("fail to start consumers for cube:" + cubeName, e); Set<Integer> rollbackStarted = successStarted.stream().map(ReplicaSet::getReplicaSetID) .collect(Collectors.toSet()); for (ReplicaSet rs : successStarted) { try { StopConsumersRequest stopRequest = new StopConsumersRequest(); stopRequest.setCube(cubeName); if (newAssign.containsKey(rs.getReplicaSetID())) { stopRequest.setRemoveData(true); } stopConsumersInReplicaSet(rs, stopRequest); rollbackStarted.remove(rs.getReplicaSetID()); } catch (IOException e1) { logger.error("fail to stop consumers for cube:" + cubeName + " replicaSet:" + rs.getReplicaSetID(), e1); } } Set<Integer> rollbackAssigned = successAssigned.stream().map(ReplicaSet::getReplicaSetID) .collect(Collectors.toSet()); for (ReplicaSet rs : successAssigned) { try { List<Partition> partitions = preAssignments.getPartitionsByReplicaSetID(rs.getReplicaSetID()); assignCubeToReplicaSet(rs, cubeName, partitions, true, true); rollbackAssigned.remove(rs.getReplicaSetID()); } catch (IOException e1) { logger.error("fail to start consumers for cube:" + cubeName + " replicaSet:" + rs.getReplicaSetID(), e1); } } Set<Node> failedReceiver = new HashSet<>(failedConvertToImmutableNodes); for (Node node : failedConvertToImmutableNodes) { try { makeCubeImmutableForReceiver(node, cubeName); failedReceiver.remove(node); } catch (IOException ioe) { logger.error("fail to make cube immutable for cube:" + cubeName + " to " + node, ioe); } } StringBuilder str = new StringBuilder(); try { str.append("FailStarted:").append(JsonUtil.writeValueAsString(rollbackStarted)).append(";"); str.append("FailAssigned:").append(JsonUtil.writeValueAsString(rollbackAssigned)).append(";"); str.append("FailRemotedPresisted:").append(JsonUtil.writeValueAsString(failedReceiver)); } catch (JsonProcessingException jpe) { logger.error("", jpe); } String failedInfo = str.toString(); if (!rollbackStarted.isEmpty()) { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_FAILED, TransactionStep.START_NEW, failedInfo, e); } else if (!rollbackAssigned.isEmpty()) { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_FAILED, TransactionStep.ASSIGN_NEW, failedInfo, e); } else if (!failedReceiver.isEmpty()) { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_FAILED, TransactionStep.MAKE_IMMUTABLE, failedInfo, e); } else { throw new ClusterStateException(cubeName, ClusterState.ROLLBACK_SUCCESS, TransactionStep.ASSIGN_NEW, failedInfo, e); } } } private  Coordinator(); @VisibleForTesting  Coordinator(StreamMetadataStore metadataStore, ReceiverAdminClient receiverClient); static Coordinator getInstance(); void start(); @VisibleForTesting void setReceiverAdminClient(ReceiverAdminClient receiverAdminClient); @VisibleForTesting void setToLeader(); @Override synchronized void assignCube(String cubeName); @Override void unAssignCube(String cubeName); @Override synchronized void reAssignCube(String cubeName, CubeAssignment assignments); @Override void segmentRemoteStoreComplete(Node receiver, String cubeName, Pair<Long, Long> segment); @Override Map<Integer, Map<String, List<Partition>>> reBalanceRecommend(); @Override synchronized void reBalance(Map<Integer, Map<String, List<Partition>>> newAssignmentsPlan); @Override void pauseConsumers(String cubeName); @Override void resumeConsumers(String cubeName); @Override void replicaSetLeaderChange(int replicaSetID, Node newLeader); StreamingCubeInfo getStreamCubeInfo(String cubeName); Map<Integer, Map<String, List<Partition>>> reBalancePlan(List<StreamingCubeInfo> allCubes,
            List<ReplicaSet> allReplicaSets); void removeCubeHDFSFiles(String cubeName); synchronized void createReplicaSet(ReplicaSet rs); synchronized void removeReplicaSet(int rsID); synchronized void addNodeToReplicaSet(Integer replicaSetID, String nodeID); synchronized void removeNodeFromReplicaSet(Integer replicaSetID, String nodeID); void stopConsumers(String cubeName); void assignCubeToReplicaSet(ReplicaSet rs, String cubeName, List<Partition> partitions,
            boolean startConsumer, boolean mustAllSucceed); ConsumerStatsResponse resumeConsumersForReceiver(final Node receiver, final ResumeConsumerRequest request); void startConsumersInReplicaSet(ReplicaSet rs, final StartConsumersRequest request); List<Node> makeCubeImmutableInReplicaSet(ReplicaSet rs, String cubeName); List<ConsumerStatsResponse> stopConsumersInReplicaSet(ReplicaSet rs, final StopConsumersRequest request); List<ConsumerStatsResponse> pauseConsumersInReplicaSet(ReplicaSet rs, final PauseConsumersRequest request); List<ConsumerStatsResponse> resumeConsumersInReplicaSet(ReplicaSet rs, final ResumeConsumerRequest request); StreamMetadataStore getStreamMetadataStore(); ExecutableManager getExecutableManager(); CubeManager getCubeManager();  }