EventHubSystemAdmin implements SystemAdmin { @Override public Map<String, SystemStreamMetadata> getSystemStreamMetadata(Set<String> streamNames) { Map<String, SystemStreamMetadata> requestedMetadata = new HashMap<>(); try { for (String streamName : streamNames) { if (!streamPartitions.containsKey(streamName)) { LOG.debug(String.format("Partition ids for Stream=%s not found", streamName)); EventHubClientManager eventHubClientManager = getOrCreateStreamEventHubClient(streamName); EventHubClient ehClient = eventHubClientManager.getEventHubClient(); CompletableFuture<EventHubRuntimeInformation> runtimeInfo = ehClient.getRuntimeInformation(); long timeoutMs = eventHubConfig.getRuntimeInfoWaitTimeMS(systemName); EventHubRuntimeInformation ehInfo = runtimeInfo.get(timeoutMs, TimeUnit.MILLISECONDS); LOG.info(String.format("Adding partition ids=%s for stream=%s. EHRuntimetInfo=%s", Arrays.toString(ehInfo.getPartitionIds()), streamName, printEventHubRuntimeInfo(ehInfo))); streamPartitions.put(streamName, ehInfo.getPartitionIds()); } String[] partitionIds = streamPartitions.get(streamName); Map<Partition, SystemStreamPartitionMetadata> sspMetadataMap = getPartitionMetadata(streamName, partitionIds); SystemStreamMetadata systemStreamMetadata = new SystemStreamMetadata(streamName, sspMetadataMap); requestedMetadata.put(streamName, systemStreamMetadata); } return requestedMetadata; } catch (Exception e) { String msg = String.format("Error while fetching EventHubRuntimeInfo for System:%s", systemName); LOG.error(msg, e); throw new SamzaException(msg, e); } } 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);  }
@Ignore("Integration Test") @Test public void testGetStreamMetadata() { EventHubSystemFactory eventHubSystemFactory = new EventHubSystemFactory(); SystemAdmin eventHubSystemAdmin = eventHubSystemFactory.getAdmin(SYSTEM_NAME, MockEventHubConfigFactory.getEventHubConfig(EventHubSystemProducer.PartitioningMethod.EVENT_HUB_HASHING)); Set<String> streams = new HashSet<>(); streams.add(STREAM_NAME1); streams.add(STREAM_NAME2); Map<String, SystemStreamMetadata> metadataMap = eventHubSystemAdmin.getSystemStreamMetadata(streams); for (String stream : streams) { Assert.assertTrue(metadataMap.containsKey(stream)); Assert.assertEquals(stream, metadataMap.get(stream).getStreamName()); Assert.assertNotNull(metadataMap.get(stream).getSystemStreamPartitionMetadata()); Map<Partition, SystemStreamMetadata.SystemStreamPartitionMetadata> partitionMetadataMap = metadataMap.get(stream).getSystemStreamPartitionMetadata(); Assert.assertTrue(partitionMetadataMap.size() >= MIN_EVENTHUB_ENTITY_PARTITION); Assert.assertTrue(partitionMetadataMap.size() <= MAX_EVENTHUB_ENTITY_PARTITION); partitionMetadataMap.forEach((partition, metadata) -> { Assert.assertEquals(EventHubSystemConsumer.START_OF_STREAM, metadata.getOldestOffset()); Assert.assertNotSame(EventHubSystemConsumer.END_OF_STREAM, metadata.getNewestOffset()); String expectedUpcomingOffset = String.valueOf(Long.parseLong(metadata.getNewestOffset()) + 1); Assert.assertEquals(expectedUpcomingOffset, metadata.getUpcomingOffset()); }); } }