@Test public void testValidateAssignment() { Map<Integer, List<Partition>> assignmentMap = new HashMap<>(); Partition p1 = new Partition(1); Partition p2 = new Partition(2); Partition p3 = new Partition(3); List<Integer> replicaSets = new ArrayList<>(); replicaSets.add(0); replicaSets.add(1); List<Partition> l1 = new ArrayList<>(); l1.add(p1); l1.add(p2); List<Partition> l2 = new ArrayList<>(); l2.add(p3); assignmentMap.put(0, l1); assignmentMap.put(1, l2); CubeAssignment cubeAssignment = new CubeAssignment("test", assignmentMap); StreamMetadataStore metadataStore = mock(StreamMetadataStore.class); when(metadataStore.getReplicaSetIDs()).thenReturn(replicaSets); StreamingV2Service streamingV2Service = new StreamingV2Service(metadataStore, null); Exception exception = null; streamingV2Service.validateAssignment(cubeAssignment); l2.add(p2); try { streamingV2Service.validateAssignment(cubeAssignment); } catch (IllegalArgumentException ill) { exception = ill; ill.printStackTrace(); } Assert.assertNotNull("Intersection detected between : 0 with 1", exception); l2.clear(); exception = null; try { streamingV2Service.validateAssignment(cubeAssignment); } catch (IllegalArgumentException ill) { exception = ill; ill.printStackTrace(); } Assert.assertNotNull("PartitionList is empty", exception); }
void validateAssignment(CubeAssignment newAssignment) { Map<Integer, List<Partition>> assignments = newAssignment.getAssignments(); Map<Integer, Set<Partition>> assignmentSet = assignments.keySet().stream().collect( Collectors.toMap(Function.identity(), x -> new HashSet<>(assignments.get(x)))); Set<Integer> inputReplicaSetIDs = assignments.keySet(); Set<Integer> allReplicaSetIDs = Sets.newHashSet(streamMetadataStore.getReplicaSetIDs()); for (Integer inputReplicaSetID : inputReplicaSetIDs) { if (!allReplicaSetIDs.contains(inputReplicaSetID)) { throw new IllegalArgumentException("The replica set id:" + inputReplicaSetID + " does not exist"); } Set<Partition> partitionSet = assignmentSet.get(inputReplicaSetID); if (partitionSet.isEmpty()) { throw new IllegalArgumentException("PartitionList is empty :" + inputReplicaSetID); } for (Map.Entry<Integer, Set<Partition>> entry : assignmentSet.entrySet()) { if (!entry.getKey().equals(inputReplicaSetID)) { Set<Partition> anotherPartitionSet = entry.getValue(); int intersection = Sets.intersection(anotherPartitionSet, partitionSet).size(); if (intersection > 0) { throw new IllegalArgumentException("Intersection detected between : " + inputReplicaSetID + " with " + entry.getKey()); } } } } }
StreamingV2Service extends BasicService { void validateAssignment(CubeAssignment newAssignment) { Map<Integer, List<Partition>> assignments = newAssignment.getAssignments(); Map<Integer, Set<Partition>> assignmentSet = assignments.keySet().stream().collect( Collectors.toMap(Function.identity(), x -> new HashSet<>(assignments.get(x)))); Set<Integer> inputReplicaSetIDs = assignments.keySet(); Set<Integer> allReplicaSetIDs = Sets.newHashSet(streamMetadataStore.getReplicaSetIDs()); for (Integer inputReplicaSetID : inputReplicaSetIDs) { if (!allReplicaSetIDs.contains(inputReplicaSetID)) { throw new IllegalArgumentException("The replica set id:" + inputReplicaSetID + " does not exist"); } Set<Partition> partitionSet = assignmentSet.get(inputReplicaSetID); if (partitionSet.isEmpty()) { throw new IllegalArgumentException("PartitionList is empty :" + inputReplicaSetID); } for (Map.Entry<Integer, Set<Partition>> entry : assignmentSet.entrySet()) { if (!entry.getKey().equals(inputReplicaSetID)) { Set<Partition> anotherPartitionSet = entry.getValue(); int intersection = Sets.intersection(anotherPartitionSet, partitionSet).size(); if (intersection > 0) { throw new IllegalArgumentException("Intersection detected between : " + inputReplicaSetID + " with " + entry.getKey()); } } } } } }
StreamingV2Service extends BasicService { void validateAssignment(CubeAssignment newAssignment) { Map<Integer, List<Partition>> assignments = newAssignment.getAssignments(); Map<Integer, Set<Partition>> assignmentSet = assignments.keySet().stream().collect( Collectors.toMap(Function.identity(), x -> new HashSet<>(assignments.get(x)))); Set<Integer> inputReplicaSetIDs = assignments.keySet(); Set<Integer> allReplicaSetIDs = Sets.newHashSet(streamMetadataStore.getReplicaSetIDs()); for (Integer inputReplicaSetID : inputReplicaSetIDs) { if (!allReplicaSetIDs.contains(inputReplicaSetID)) { throw new IllegalArgumentException("The replica set id:" + inputReplicaSetID + " does not exist"); } Set<Partition> partitionSet = assignmentSet.get(inputReplicaSetID); if (partitionSet.isEmpty()) { throw new IllegalArgumentException("PartitionList is empty :" + inputReplicaSetID); } for (Map.Entry<Integer, Set<Partition>> entry : assignmentSet.entrySet()) { if (!entry.getKey().equals(inputReplicaSetID)) { Set<Partition> anotherPartitionSet = entry.getValue(); int intersection = Sets.intersection(anotherPartitionSet, partitionSet).size(); if (intersection > 0) { throw new IllegalArgumentException("Intersection detected between : " + inputReplicaSetID + " with " + entry.getKey()); } } } } } StreamingV2Service();   StreamingV2Service(StreamMetadataStore metadataStore, ReceiverAdminClient adminClient); }
StreamingV2Service extends BasicService { void validateAssignment(CubeAssignment newAssignment) { Map<Integer, List<Partition>> assignments = newAssignment.getAssignments(); Map<Integer, Set<Partition>> assignmentSet = assignments.keySet().stream().collect( Collectors.toMap(Function.identity(), x -> new HashSet<>(assignments.get(x)))); Set<Integer> inputReplicaSetIDs = assignments.keySet(); Set<Integer> allReplicaSetIDs = Sets.newHashSet(streamMetadataStore.getReplicaSetIDs()); for (Integer inputReplicaSetID : inputReplicaSetIDs) { if (!allReplicaSetIDs.contains(inputReplicaSetID)) { throw new IllegalArgumentException("The replica set id:" + inputReplicaSetID + " does not exist"); } Set<Partition> partitionSet = assignmentSet.get(inputReplicaSetID); if (partitionSet.isEmpty()) { throw new IllegalArgumentException("PartitionList is empty :" + inputReplicaSetID); } for (Map.Entry<Integer, Set<Partition>> entry : assignmentSet.entrySet()) { if (!entry.getKey().equals(inputReplicaSetID)) { Set<Partition> anotherPartitionSet = entry.getValue(); int intersection = Sets.intersection(anotherPartitionSet, partitionSet).size(); if (intersection > 0) { throw new IllegalArgumentException("Intersection detected between : " + inputReplicaSetID + " with " + entry.getKey()); } } } } } StreamingV2Service();   StreamingV2Service(StreamMetadataStore metadataStore, ReceiverAdminClient adminClient); List<StreamingSourceConfig> listAllStreamingConfigs(final String table); List<StreamingSourceConfig> getStreamingConfigs(final String table, final Integer limit, final Integer offset); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN + " or hasPermission(#project, 'ADMINISTRATION')") StreamingSourceConfig createStreamingConfig(StreamingSourceConfig config, ProjectInstance project); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) StreamingSourceConfig updateStreamingConfig(StreamingSourceConfig config); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) void dropStreamingConfig(StreamingSourceConfig config); String getParserTemplate(final int sourceType, StreamingSourceConfig config); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN + " or hasPermission(#cube, 'ADMINISTRATION') or hasPermission(#cube, 'MANAGEMENT')") List<CubeAssignment> getStreamingCubeAssignments(final CubeInstance cube); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) Map<Integer, Map<String, List<Partition>>> getStreamingReplicaSetAssignments(Integer replicaSetID); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) Map<Integer, Map<String, List<Partition>>> reBalancePlan(); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) void reBalance(Map<Integer, Map<String, List<Partition>>> reBalancePlan); List<String> getStreamingCubes(); StreamingCubeConsumeState getStreamingCubeConsumeState(String cubeName); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN + " or hasPermission(#cube, 'ADMINISTRATION') or hasPermission(#cube, 'MANAGEMENT')") void assignCube(CubeInstance cube); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN + " or hasPermission(#cube, 'ADMINISTRATION') or hasPermission(#cube, 'MANAGEMENT')") void unAssignCube(CubeInstance cube); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) void reAssignCube(String cubeName, CubeAssignment newAssignment); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN + " or hasPermission(#cube, 'ADMINISTRATION') or hasPermission(#cube, 'MANAGEMENT')") void pauseConsumers(CubeInstance cube); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN + " or hasPermission(#cube, 'ADMINISTRATION') or hasPermission(#cube, 'MANAGEMENT')") void resumeConsumers(CubeInstance cube); StreamingSourceConfigManager getStreamingManagerV2(); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) void removeCubeAssignment(); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) List<Node> getReceivers(); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) void removeReceiver(Node receiver); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) void createReplicaSet(ReplicaSet rs); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) void removeReplicaSet(int rsID); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) void addNodeToReplicaSet(Integer replicaSetID, String nodeID); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) void removeNodeFromReplicaSet(Integer replicaSetID, String nodeID); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) List<ReplicaSet> getReplicaSets(); ReceiverStats getReceiverStats(Node receiver); ReceiverCubeStats getReceiverCubeStats(Node receiver, String cubeName); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN + " or hasPermission(#cube, 'ADMINISTRATION') or hasPermission(#cube, 'MANAGEMENT') or hasPermission(#cube.getProjectInstance(), 'MANAGEMENT') or hasPermission(#cube.getProjectInstance(), 'ADMINISTRATION')") CubeRealTimeState getCubeRealTimeState(CubeInstance cube); ClusterState getClusterState(); }
StreamingV2Service extends BasicService { void validateAssignment(CubeAssignment newAssignment) { Map<Integer, List<Partition>> assignments = newAssignment.getAssignments(); Map<Integer, Set<Partition>> assignmentSet = assignments.keySet().stream().collect( Collectors.toMap(Function.identity(), x -> new HashSet<>(assignments.get(x)))); Set<Integer> inputReplicaSetIDs = assignments.keySet(); Set<Integer> allReplicaSetIDs = Sets.newHashSet(streamMetadataStore.getReplicaSetIDs()); for (Integer inputReplicaSetID : inputReplicaSetIDs) { if (!allReplicaSetIDs.contains(inputReplicaSetID)) { throw new IllegalArgumentException("The replica set id:" + inputReplicaSetID + " does not exist"); } Set<Partition> partitionSet = assignmentSet.get(inputReplicaSetID); if (partitionSet.isEmpty()) { throw new IllegalArgumentException("PartitionList is empty :" + inputReplicaSetID); } for (Map.Entry<Integer, Set<Partition>> entry : assignmentSet.entrySet()) { if (!entry.getKey().equals(inputReplicaSetID)) { Set<Partition> anotherPartitionSet = entry.getValue(); int intersection = Sets.intersection(anotherPartitionSet, partitionSet).size(); if (intersection > 0) { throw new IllegalArgumentException("Intersection detected between : " + inputReplicaSetID + " with " + entry.getKey()); } } } } } StreamingV2Service();   StreamingV2Service(StreamMetadataStore metadataStore, ReceiverAdminClient adminClient); List<StreamingSourceConfig> listAllStreamingConfigs(final String table); List<StreamingSourceConfig> getStreamingConfigs(final String table, final Integer limit, final Integer offset); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN + " or hasPermission(#project, 'ADMINISTRATION')") StreamingSourceConfig createStreamingConfig(StreamingSourceConfig config, ProjectInstance project); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) StreamingSourceConfig updateStreamingConfig(StreamingSourceConfig config); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) void dropStreamingConfig(StreamingSourceConfig config); String getParserTemplate(final int sourceType, StreamingSourceConfig config); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN + " or hasPermission(#cube, 'ADMINISTRATION') or hasPermission(#cube, 'MANAGEMENT')") List<CubeAssignment> getStreamingCubeAssignments(final CubeInstance cube); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) Map<Integer, Map<String, List<Partition>>> getStreamingReplicaSetAssignments(Integer replicaSetID); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) Map<Integer, Map<String, List<Partition>>> reBalancePlan(); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) void reBalance(Map<Integer, Map<String, List<Partition>>> reBalancePlan); List<String> getStreamingCubes(); StreamingCubeConsumeState getStreamingCubeConsumeState(String cubeName); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN + " or hasPermission(#cube, 'ADMINISTRATION') or hasPermission(#cube, 'MANAGEMENT')") void assignCube(CubeInstance cube); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN + " or hasPermission(#cube, 'ADMINISTRATION') or hasPermission(#cube, 'MANAGEMENT')") void unAssignCube(CubeInstance cube); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) void reAssignCube(String cubeName, CubeAssignment newAssignment); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN + " or hasPermission(#cube, 'ADMINISTRATION') or hasPermission(#cube, 'MANAGEMENT')") void pauseConsumers(CubeInstance cube); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN + " or hasPermission(#cube, 'ADMINISTRATION') or hasPermission(#cube, 'MANAGEMENT')") void resumeConsumers(CubeInstance cube); StreamingSourceConfigManager getStreamingManagerV2(); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) void removeCubeAssignment(); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) List<Node> getReceivers(); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) void removeReceiver(Node receiver); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) void createReplicaSet(ReplicaSet rs); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) void removeReplicaSet(int rsID); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) void addNodeToReplicaSet(Integer replicaSetID, String nodeID); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) void removeNodeFromReplicaSet(Integer replicaSetID, String nodeID); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN) List<ReplicaSet> getReplicaSets(); ReceiverStats getReceiverStats(Node receiver); ReceiverCubeStats getReceiverCubeStats(Node receiver, String cubeName); @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN + " or hasPermission(#cube, 'ADMINISTRATION') or hasPermission(#cube, 'MANAGEMENT') or hasPermission(#cube.getProjectInstance(), 'MANAGEMENT') or hasPermission(#cube.getProjectInstance(), 'ADMINISTRATION')") CubeRealTimeState getCubeRealTimeState(CubeInstance cube); ClusterState getClusterState();  }