@Test public void testSegmentToMergeWithIsUnclaimed() { int expectedInvocations = 1; eventProcessors.clear(); ClientProcessor splitClientProcessor = mock(ClientProcessor.class); when(splitClientProcessor.clientId()).thenReturn(CLIENT_NAME_OF_SPLIT); EventProcessorInfo eventProcessorToSplit = EventProcessorInfo.newBuilder() .setProcessorName(PROCESSOR_NAME_TO_SPLIT) .addAllSegmentStatus( Arrays.asList( createSegmentInfo(0, 4), createSegmentInfo(1, 2) ) ) .build(); when(splitClientProcessor.eventProcessorInfo()).thenReturn(eventProcessorToSplit); eventProcessors.add(splitClientProcessor); testSubject.mergeSegment(CONTEXT, CLIENTS, PROCESSOR_NAME_TO_SPLIT); ArgumentCaptor<Object> argumentCaptor = ArgumentCaptor.forClass(Object.class); verify(applicationEventPublisher, times(expectedInvocations)).publishEvent(argumentCaptor.capture()); List<Object> publishedEvents = argumentCaptor.getAllValues(); MergeSegmentRequest mergeRequest = (MergeSegmentRequest) publishedEvents.get(0); assertFalse(mergeRequest.isProxied()); assertEquals(CLIENT_NAME_OF_SPLIT, mergeRequest.getClientId()); assertEquals(PROCESSOR_NAME_TO_SPLIT, mergeRequest.getProcessorName()); assertEquals(SEGMENT_ID_TO_MERGE, mergeRequest.getSegmentId()); }
public void mergeSegment(String context, List<String> clientIds, String processorName) { Map<ClientSegmentPair, SegmentStatus> clientToTracker = buildClientToTrackerMap(clientIds, processorName, REVERSE_ORDER); SegmentStatus smallestSegment = clientToTracker.values().stream() .max(Comparator.comparingInt(SegmentStatus::getOnePartOf)) .orElseThrow(() -> new IllegalArgumentException( "No segments found for processor name [" + processorName + "]" )); int segmentToMerge = deduceSegmentToMerge(smallestSegment); Optional<String> clientIdOwningSegmentToMerge = getClientIdForSegment(clientToTracker, segmentToMerge); clientIdOwningSegmentToMerge.ifPresent( clientId -> { clientIds.stream() .filter(client -> !client.equals(clientId)) .forEach(client -> releaseSegment(context, client, processorName, smallestSegment.getSegmentId())); } ); if (clientIdOwningSegmentToMerge.isPresent()) { applicationEventPublisher.publishEvent(new MergeSegmentRequest( NOT_PROXIED, context, clientIdOwningSegmentToMerge.get(), processorName, segmentToMerge )); } else { String clientOwningSmallestSegment = getClientIdForSegment(clientToTracker, smallestSegment.getSegmentId()) .orElseThrow(() -> new IllegalArgumentException( "Attempt to merge segments [" + segmentToMerge + "] and [" + smallestSegment.getSegmentId() + "] failed") ); applicationEventPublisher.publishEvent(new MergeSegmentRequest( NOT_PROXIED, context, clientOwningSmallestSegment, processorName, smallestSegment.getSegmentId() )); } }
ProcessorEventPublisher { public void mergeSegment(String context, List<String> clientIds, String processorName) { Map<ClientSegmentPair, SegmentStatus> clientToTracker = buildClientToTrackerMap(clientIds, processorName, REVERSE_ORDER); SegmentStatus smallestSegment = clientToTracker.values().stream() .max(Comparator.comparingInt(SegmentStatus::getOnePartOf)) .orElseThrow(() -> new IllegalArgumentException( "No segments found for processor name [" + processorName + "]" )); int segmentToMerge = deduceSegmentToMerge(smallestSegment); Optional<String> clientIdOwningSegmentToMerge = getClientIdForSegment(clientToTracker, segmentToMerge); clientIdOwningSegmentToMerge.ifPresent( clientId -> { clientIds.stream() .filter(client -> !client.equals(clientId)) .forEach(client -> releaseSegment(context, client, processorName, smallestSegment.getSegmentId())); } ); if (clientIdOwningSegmentToMerge.isPresent()) { applicationEventPublisher.publishEvent(new MergeSegmentRequest( NOT_PROXIED, context, clientIdOwningSegmentToMerge.get(), processorName, segmentToMerge )); } else { String clientOwningSmallestSegment = getClientIdForSegment(clientToTracker, smallestSegment.getSegmentId()) .orElseThrow(() -> new IllegalArgumentException( "Attempt to merge segments [" + segmentToMerge + "] and [" + smallestSegment.getSegmentId() + "] failed") ); applicationEventPublisher.publishEvent(new MergeSegmentRequest( NOT_PROXIED, context, clientOwningSmallestSegment, processorName, smallestSegment.getSegmentId() )); } } }
ProcessorEventPublisher { public void mergeSegment(String context, List<String> clientIds, String processorName) { Map<ClientSegmentPair, SegmentStatus> clientToTracker = buildClientToTrackerMap(clientIds, processorName, REVERSE_ORDER); SegmentStatus smallestSegment = clientToTracker.values().stream() .max(Comparator.comparingInt(SegmentStatus::getOnePartOf)) .orElseThrow(() -> new IllegalArgumentException( "No segments found for processor name [" + processorName + "]" )); int segmentToMerge = deduceSegmentToMerge(smallestSegment); Optional<String> clientIdOwningSegmentToMerge = getClientIdForSegment(clientToTracker, segmentToMerge); clientIdOwningSegmentToMerge.ifPresent( clientId -> { clientIds.stream() .filter(client -> !client.equals(clientId)) .forEach(client -> releaseSegment(context, client, processorName, smallestSegment.getSegmentId())); } ); if (clientIdOwningSegmentToMerge.isPresent()) { applicationEventPublisher.publishEvent(new MergeSegmentRequest( NOT_PROXIED, context, clientIdOwningSegmentToMerge.get(), processorName, segmentToMerge )); } else { String clientOwningSmallestSegment = getClientIdForSegment(clientToTracker, smallestSegment.getSegmentId()) .orElseThrow(() -> new IllegalArgumentException( "Attempt to merge segments [" + segmentToMerge + "] and [" + smallestSegment.getSegmentId() + "] failed") ); applicationEventPublisher.publishEvent(new MergeSegmentRequest( NOT_PROXIED, context, clientOwningSmallestSegment, processorName, smallestSegment.getSegmentId() )); } } ProcessorEventPublisher(PlatformService platformService,
                                   ApplicationEventPublisher applicationEventPublisher,
                                   ClientProcessors clientProcessors); }
ProcessorEventPublisher { public void mergeSegment(String context, List<String> clientIds, String processorName) { Map<ClientSegmentPair, SegmentStatus> clientToTracker = buildClientToTrackerMap(clientIds, processorName, REVERSE_ORDER); SegmentStatus smallestSegment = clientToTracker.values().stream() .max(Comparator.comparingInt(SegmentStatus::getOnePartOf)) .orElseThrow(() -> new IllegalArgumentException( "No segments found for processor name [" + processorName + "]" )); int segmentToMerge = deduceSegmentToMerge(smallestSegment); Optional<String> clientIdOwningSegmentToMerge = getClientIdForSegment(clientToTracker, segmentToMerge); clientIdOwningSegmentToMerge.ifPresent( clientId -> { clientIds.stream() .filter(client -> !client.equals(clientId)) .forEach(client -> releaseSegment(context, client, processorName, smallestSegment.getSegmentId())); } ); if (clientIdOwningSegmentToMerge.isPresent()) { applicationEventPublisher.publishEvent(new MergeSegmentRequest( NOT_PROXIED, context, clientIdOwningSegmentToMerge.get(), processorName, segmentToMerge )); } else { String clientOwningSmallestSegment = getClientIdForSegment(clientToTracker, smallestSegment.getSegmentId()) .orElseThrow(() -> new IllegalArgumentException( "Attempt to merge segments [" + segmentToMerge + "] and [" + smallestSegment.getSegmentId() + "] failed") ); applicationEventPublisher.publishEvent(new MergeSegmentRequest( NOT_PROXIED, context, clientOwningSmallestSegment, processorName, smallestSegment.getSegmentId() )); } } ProcessorEventPublisher(PlatformService platformService,
                                   ApplicationEventPublisher applicationEventPublisher,
                                   ClientProcessors clientProcessors); @PostConstruct void init(); void pauseProcessorRequest(String context, String clientId, String processorName); void startProcessorRequest(String context, String clientId, String processorName); void releaseSegment(String context, String clientId, String processorName, int segmentId); void splitSegment(String context, List<String> clientIds, String processorName); void mergeSegment(String context, List<String> clientIds, String processorName); }
ProcessorEventPublisher { public void mergeSegment(String context, List<String> clientIds, String processorName) { Map<ClientSegmentPair, SegmentStatus> clientToTracker = buildClientToTrackerMap(clientIds, processorName, REVERSE_ORDER); SegmentStatus smallestSegment = clientToTracker.values().stream() .max(Comparator.comparingInt(SegmentStatus::getOnePartOf)) .orElseThrow(() -> new IllegalArgumentException( "No segments found for processor name [" + processorName + "]" )); int segmentToMerge = deduceSegmentToMerge(smallestSegment); Optional<String> clientIdOwningSegmentToMerge = getClientIdForSegment(clientToTracker, segmentToMerge); clientIdOwningSegmentToMerge.ifPresent( clientId -> { clientIds.stream() .filter(client -> !client.equals(clientId)) .forEach(client -> releaseSegment(context, client, processorName, smallestSegment.getSegmentId())); } ); if (clientIdOwningSegmentToMerge.isPresent()) { applicationEventPublisher.publishEvent(new MergeSegmentRequest( NOT_PROXIED, context, clientIdOwningSegmentToMerge.get(), processorName, segmentToMerge )); } else { String clientOwningSmallestSegment = getClientIdForSegment(clientToTracker, smallestSegment.getSegmentId()) .orElseThrow(() -> new IllegalArgumentException( "Attempt to merge segments [" + segmentToMerge + "] and [" + smallestSegment.getSegmentId() + "] failed") ); applicationEventPublisher.publishEvent(new MergeSegmentRequest( NOT_PROXIED, context, clientOwningSmallestSegment, processorName, smallestSegment.getSegmentId() )); } } ProcessorEventPublisher(PlatformService platformService,
                                   ApplicationEventPublisher applicationEventPublisher,
                                   ClientProcessors clientProcessors); @PostConstruct void init(); void pauseProcessorRequest(String context, String clientId, String processorName); void startProcessorRequest(String context, String clientId, String processorName); void releaseSegment(String context, String clientId, String processorName, int segmentId); void splitSegment(String context, List<String> clientIds, String processorName); void mergeSegment(String context, List<String> clientIds, String processorName);  }