EventHubSystemAdmin implements SystemAdmin { EventHubClientManager getOrCreateStreamEventHubClient(String streamName) { if (!eventHubClients.containsKey(streamName)) { LOG.info(String.format("Creating EventHubClient for Stream=%s", streamName)); EventHubClientManager eventHubClientManager = eventHubClientManagerFactory .getEventHubClientManager(systemName, streamName, eventHubConfig); eventHubClientManager.init(); eventHubClients.put(streamName, eventHubClientManager); } return eventHubClients.get(streamName); } EventHubSystemAdmin(String systemName, EventHubConfig eventHubConfig,
                             EventHubClientManagerFactory eventHubClientManagerFactory); @Override void stop(); @Override Map<SystemStreamPartition, String> getOffsetsAfter(Map<SystemStreamPartition, String> offsets); @Override Map<String, SystemStreamMetadata> getSystemStreamMetadata(Set<String> streamNames); @Override String resolveStartpointToOffset(SystemStreamPartition systemStreamPartition, Startpoint startpoint); static Integer compareOffsets(String offset1, String offset2); @Override Integer offsetComparator(String offset1, String offset2);  }
@Test public void testStartpointResolverShouldResolveTheStartpointTimestampToCorrectOffset() throws EventHubException { EventHubSystemAdmin mockEventHubSystemAdmin = Mockito.mock(EventHubSystemAdmin.class); EventHubConfig eventHubConfig = Mockito.mock(EventHubConfig.class); SystemStreamPartition systemStreamPartition = new SystemStreamPartition("test-system", "test-stream", new Partition(0)); String mockedOffsetToReturn = "100"; EventHubClientManager mockEventHubClientManager = Mockito.mock(EventHubClientManager.class); EventHubClient mockEventHubClient = Mockito.mock(EventHubClient.class); PartitionReceiver mockPartitionReceiver = Mockito.mock(PartitionReceiver.class); EventData mockEventData = Mockito.mock(EventData.class); EventData.SystemProperties mockSystemProperties = Mockito.mock(EventData.SystemProperties.class); Mockito.when(mockEventHubSystemAdmin.getOrCreateStreamEventHubClient("test-stream")).thenReturn(mockEventHubClientManager); Mockito.when(mockEventHubClientManager.getEventHubClient()).thenReturn(mockEventHubClient); Mockito.when(mockEventHubClient.createReceiverSync(Mockito.anyString(), Mockito.anyString(), Mockito.any())).thenReturn(mockPartitionReceiver); Mockito.when(mockPartitionReceiver.receiveSync(1)).thenReturn(Arrays.asList(mockEventData)); Mockito.when(mockEventData.getSystemProperties()).thenReturn(mockSystemProperties); Mockito.when(mockSystemProperties.getOffset()).thenReturn(mockedOffsetToReturn); EventHubSamzaOffsetResolver resolver = new EventHubSamzaOffsetResolver(mockEventHubSystemAdmin, eventHubConfig); String resolvedOffset = resolver.visit(systemStreamPartition, new StartpointTimestamp(100L)); Assert.assertEquals(mockedOffsetToReturn, resolvedOffset); }