@Test public void testSubmitUnfulfilledRequestEvents() throws IOException, NoSuchMethodException, InvocationTargetException, IllegalAccessException { SourceState state = new SourceState(); state.setProp(ConfigurationKeys.SOURCE_FILEBASED_FS_URI, "file: state.setProp(ConfigurationKeys.WRITER_FILE_SYSTEM_URI, "file: state.setProp(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, "/target/dir"); state.setProp(DatasetUtils.DATASET_PROFILE_CLASS_KEY, TestCopyablePartitionableDatasedFinder.class.getCanonicalName()); state.setProp(CopySource.MAX_CONCURRENT_LISTING_SERVICES, 2); state.setProp(CopyConfiguration.MAX_COPY_PREFIX + ".size", "50"); state.setProp(CopyConfiguration.MAX_COPY_PREFIX + ".copyEntities", 2); state.setProp(CopyConfiguration.STORE_REJECTED_REQUESTS_KEY, RequestAllocatorConfig.StoreRejectedRequestsConfig.ALL.name().toLowerCase()); state.setProp(ConfigurationKeys.METRICS_CUSTOM_BUILDERS, "org.apache.gobblin.metrics.ConsoleEventReporterFactory"); CopySource source = new CopySource(); final FileSystem sourceFs = HadoopUtils.getSourceFileSystem(state); final FileSystem targetFs = HadoopUtils.getWriterFileSystem(state, 1, 0); int maxThreads = state .getPropAsInt(CopySource.MAX_CONCURRENT_LISTING_SERVICES, CopySource.DEFAULT_MAX_CONCURRENT_LISTING_SERVICES); final CopyConfiguration copyConfiguration = CopyConfiguration.builder(targetFs, state.getProperties()).build(); MetricContext metricContext = Instrumented.getMetricContext(state, CopySource.class); EventSubmitter eventSubmitter = new EventSubmitter.Builder(metricContext, CopyConfiguration.COPY_PREFIX).build(); DatasetsFinder<CopyableDatasetBase> datasetFinder = DatasetUtils .instantiateDatasetFinder(state.getProperties(), sourceFs, CopySource.DEFAULT_DATASET_PROFILE_CLASS_KEY, eventSubmitter, state); IterableDatasetFinder<CopyableDatasetBase> iterableDatasetFinder = datasetFinder instanceof IterableDatasetFinder ? (IterableDatasetFinder<CopyableDatasetBase>) datasetFinder : new IterableDatasetFinderImpl<>(datasetFinder); Iterator<CopyableDatasetRequestor> requestorIteratorWithNulls = Iterators .transform(iterableDatasetFinder.getDatasetsIterator(), new CopyableDatasetRequestor.Factory(targetFs, copyConfiguration, log)); Iterator<CopyableDatasetRequestor> requestorIterator = Iterators.filter(requestorIteratorWithNulls, Predicates.<CopyableDatasetRequestor>notNull()); Method m = CopySource.class.getDeclaredMethod("createRequestAllocator", CopyConfiguration.class, int.class); m.setAccessible(true); PriorityIterableBasedRequestAllocator<FileSet<CopyEntity>> allocator = (PriorityIterableBasedRequestAllocator<FileSet<CopyEntity>>) m.invoke(source, copyConfiguration, maxThreads); Iterator<FileSet<CopyEntity>> prioritizedFileSets = allocator.allocateRequests(requestorIterator, copyConfiguration.getMaxToCopy()); List<FileSet<CopyEntity>> fileSetList = allocator.getRequestsExceedingAvailableResourcePool(); Assert.assertEquals(fileSetList.size(), 2); FileSet<CopyEntity> fileSet = fileSetList.get(0); Assert.assertEquals(fileSet.getDataset().getUrn(), "/test"); Assert.assertEquals(fileSet.getTotalEntities(), 5); Assert.assertEquals(fileSet.getTotalSizeInBytes(), 50); fileSet = fileSetList.get(1); Assert.assertEquals(fileSet.getDataset().getUrn(), "/test"); Assert.assertEquals(fileSet.getTotalEntities(), 5); Assert.assertEquals(fileSet.getTotalSizeInBytes(), 50); }
private void submitUnfulfilledRequestEvents(RequestAllocator<FileSet<CopyEntity>> allocator) { if (PriorityIterableBasedRequestAllocator.class.isAssignableFrom(allocator.getClass())) { PriorityIterableBasedRequestAllocator<FileSet<CopyEntity>> priorityIterableBasedRequestAllocator = (PriorityIterableBasedRequestAllocator<FileSet<CopyEntity>>) allocator; submitUnfulfilledRequestEventsHelper( priorityIterableBasedRequestAllocator.getRequestsExceedingAvailableResourcePool(), REQUESTS_EXCEEDING_AVAILABLE_RESOURCE_POOL_EVENT_NAME); submitUnfulfilledRequestEventsHelper( priorityIterableBasedRequestAllocator.getRequestsRejectedDueToInsufficientEviction(), REQUESTS_REJECTED_DUE_TO_INSUFFICIENT_EVICTION_EVENT_NAME); submitUnfulfilledRequestEventsHelper(priorityIterableBasedRequestAllocator.getRequestsRejectedWithLowPriority(), REQUESTS_REJECTED_WITH_LOW_PRIORITY_EVENT_NAME); submitUnfulfilledRequestEventsHelper(priorityIterableBasedRequestAllocator.getRequestsDropped(), REQUESTS_DROPPED_EVENT_NAME); } }
CopySource extends AbstractSource<String, FileAwareInputStream> { private void submitUnfulfilledRequestEvents(RequestAllocator<FileSet<CopyEntity>> allocator) { if (PriorityIterableBasedRequestAllocator.class.isAssignableFrom(allocator.getClass())) { PriorityIterableBasedRequestAllocator<FileSet<CopyEntity>> priorityIterableBasedRequestAllocator = (PriorityIterableBasedRequestAllocator<FileSet<CopyEntity>>) allocator; submitUnfulfilledRequestEventsHelper( priorityIterableBasedRequestAllocator.getRequestsExceedingAvailableResourcePool(), REQUESTS_EXCEEDING_AVAILABLE_RESOURCE_POOL_EVENT_NAME); submitUnfulfilledRequestEventsHelper( priorityIterableBasedRequestAllocator.getRequestsRejectedDueToInsufficientEviction(), REQUESTS_REJECTED_DUE_TO_INSUFFICIENT_EVICTION_EVENT_NAME); submitUnfulfilledRequestEventsHelper(priorityIterableBasedRequestAllocator.getRequestsRejectedWithLowPriority(), REQUESTS_REJECTED_WITH_LOW_PRIORITY_EVENT_NAME); submitUnfulfilledRequestEventsHelper(priorityIterableBasedRequestAllocator.getRequestsDropped(), REQUESTS_DROPPED_EVENT_NAME); } } }
CopySource extends AbstractSource<String, FileAwareInputStream> { private void submitUnfulfilledRequestEvents(RequestAllocator<FileSet<CopyEntity>> allocator) { if (PriorityIterableBasedRequestAllocator.class.isAssignableFrom(allocator.getClass())) { PriorityIterableBasedRequestAllocator<FileSet<CopyEntity>> priorityIterableBasedRequestAllocator = (PriorityIterableBasedRequestAllocator<FileSet<CopyEntity>>) allocator; submitUnfulfilledRequestEventsHelper( priorityIterableBasedRequestAllocator.getRequestsExceedingAvailableResourcePool(), REQUESTS_EXCEEDING_AVAILABLE_RESOURCE_POOL_EVENT_NAME); submitUnfulfilledRequestEventsHelper( priorityIterableBasedRequestAllocator.getRequestsRejectedDueToInsufficientEviction(), REQUESTS_REJECTED_DUE_TO_INSUFFICIENT_EVICTION_EVENT_NAME); submitUnfulfilledRequestEventsHelper(priorityIterableBasedRequestAllocator.getRequestsRejectedWithLowPriority(), REQUESTS_REJECTED_WITH_LOW_PRIORITY_EVENT_NAME); submitUnfulfilledRequestEventsHelper(priorityIterableBasedRequestAllocator.getRequestsDropped(), REQUESTS_DROPPED_EVENT_NAME); } }  }
CopySource extends AbstractSource<String, FileAwareInputStream> { private void submitUnfulfilledRequestEvents(RequestAllocator<FileSet<CopyEntity>> allocator) { if (PriorityIterableBasedRequestAllocator.class.isAssignableFrom(allocator.getClass())) { PriorityIterableBasedRequestAllocator<FileSet<CopyEntity>> priorityIterableBasedRequestAllocator = (PriorityIterableBasedRequestAllocator<FileSet<CopyEntity>>) allocator; submitUnfulfilledRequestEventsHelper( priorityIterableBasedRequestAllocator.getRequestsExceedingAvailableResourcePool(), REQUESTS_EXCEEDING_AVAILABLE_RESOURCE_POOL_EVENT_NAME); submitUnfulfilledRequestEventsHelper( priorityIterableBasedRequestAllocator.getRequestsRejectedDueToInsufficientEviction(), REQUESTS_REJECTED_DUE_TO_INSUFFICIENT_EVICTION_EVENT_NAME); submitUnfulfilledRequestEventsHelper(priorityIterableBasedRequestAllocator.getRequestsRejectedWithLowPriority(), REQUESTS_REJECTED_WITH_LOW_PRIORITY_EVENT_NAME); submitUnfulfilledRequestEventsHelper(priorityIterableBasedRequestAllocator.getRequestsDropped(), REQUESTS_DROPPED_EVENT_NAME); } }  @Override List<WorkUnit> getWorkunits(final SourceState state); @Override Extractor<String, FileAwareInputStream> getExtractor(WorkUnitState state); @Override void shutdown(SourceState state); static void setWorkUnitGuid(State state, Guid guid); static Optional<Guid> getWorkUnitGuid(State state); static void serializeCopyEntity(State state, CopyEntity copyEntity); static Class<?> getCopyEntityClass(State state); static CopyEntity deserializeCopyEntity(State state); static void serializeCopyableDataset(State state, CopyableDatasetMetadata copyableDataset); static CopyableDatasetMetadata deserializeCopyableDataset(State state); }
CopySource extends AbstractSource<String, FileAwareInputStream> { private void submitUnfulfilledRequestEvents(RequestAllocator<FileSet<CopyEntity>> allocator) { if (PriorityIterableBasedRequestAllocator.class.isAssignableFrom(allocator.getClass())) { PriorityIterableBasedRequestAllocator<FileSet<CopyEntity>> priorityIterableBasedRequestAllocator = (PriorityIterableBasedRequestAllocator<FileSet<CopyEntity>>) allocator; submitUnfulfilledRequestEventsHelper( priorityIterableBasedRequestAllocator.getRequestsExceedingAvailableResourcePool(), REQUESTS_EXCEEDING_AVAILABLE_RESOURCE_POOL_EVENT_NAME); submitUnfulfilledRequestEventsHelper( priorityIterableBasedRequestAllocator.getRequestsRejectedDueToInsufficientEviction(), REQUESTS_REJECTED_DUE_TO_INSUFFICIENT_EVICTION_EVENT_NAME); submitUnfulfilledRequestEventsHelper(priorityIterableBasedRequestAllocator.getRequestsRejectedWithLowPriority(), REQUESTS_REJECTED_WITH_LOW_PRIORITY_EVENT_NAME); submitUnfulfilledRequestEventsHelper(priorityIterableBasedRequestAllocator.getRequestsDropped(), REQUESTS_DROPPED_EVENT_NAME); } }  @Override List<WorkUnit> getWorkunits(final SourceState state); @Override Extractor<String, FileAwareInputStream> getExtractor(WorkUnitState state); @Override void shutdown(SourceState state); static void setWorkUnitGuid(State state, Guid guid); static Optional<Guid> getWorkUnitGuid(State state); static void serializeCopyEntity(State state, CopyEntity copyEntity); static Class<?> getCopyEntityClass(State state); static CopyEntity deserializeCopyEntity(State state); static void serializeCopyableDataset(State state, CopyableDatasetMetadata copyableDataset); static CopyableDatasetMetadata deserializeCopyableDataset(State state); static final String DEFAULT_DATASET_PROFILE_CLASS_KEY; static final String SERIALIZED_COPYABLE_FILE; static final String COPY_ENTITY_CLASS; static final String SERIALIZED_COPYABLE_DATASET; static final String WORK_UNIT_GUID; static final String MAX_CONCURRENT_LISTING_SERVICES; static final int DEFAULT_MAX_CONCURRENT_LISTING_SERVICES; static final String MAX_FILES_COPIED_KEY; static final String SIMULATE; static final String MAX_SIZE_MULTI_WORKUNITS; static final String MAX_WORK_UNITS_PER_BIN; static final String REQUESTS_EXCEEDING_AVAILABLE_RESOURCE_POOL_EVENT_NAME; static final String REQUESTS_DROPPED_EVENT_NAME; static final String REQUESTS_REJECTED_DUE_TO_INSUFFICIENT_EVICTION_EVENT_NAME; static final String REQUESTS_REJECTED_WITH_LOW_PRIORITY_EVENT_NAME; static final String FILESET_NAME; static final String FILESET_TOTAL_ENTITIES; static final String FILESET_TOTAL_SIZE_IN_BYTES; static final String SCHEMA_CHECK_ENABLED; static final String DATASET_STAGING_DIR_PATH; static final String DATASET_STAGING_PATH; final static boolean DEFAULT_SCHEMA_CHECK_ENABLED; public MetricContext metricContext; public EventSubmitter eventSubmitter; }