DatastreamResources extends CollectionResourceTemplate<String, Datastream> { @SuppressWarnings("deprecated") @Override public List<Datastream> getAll(@Context PagingContext pagingContext) { try { LOG.info("Get all datastreams called with paging context {}", pagingContext); _dynamicMetricsManager.createOrUpdateMeter(CLASS_NAME, GET_ALL_CALL, 1); Instant startTime = Instant.now(); List<Datastream> ret = RestliUtils.withPaging(_store.getAllDatastreams(), pagingContext) .map(_store::getDatastream) .filter(Objects::nonNull) .collect(Collectors.toList()); LOG.info("Get all datastreams call took {} ms", Duration.between(startTime, Instant.now()).toMillis()); LOG.debug("Result collected for getAll {}", ret); return ret; } catch (Exception e) { _dynamicMetricsManager.createOrUpdateMeter(CLASS_NAME, CALL_ERROR, 1); _errorLogger.logAndThrowRestLiServiceException(HttpStatus.S_500_INTERNAL_SERVER_ERROR, "Get all datastreams failed.", e); } return Collections.emptyList(); } DatastreamResources(DatastreamServer datastreamServer);  DatastreamResources(DatastreamStore store, Coordinator coordinator); @Override BatchUpdateResult<String, Datastream> batchUpdate(final BatchUpdateRequest<String, Datastream> entities); @Override UpdateResponse update(String key, Datastream datastream); @Action(name = "pause", resourceLevel = ResourceLevel.ENTITY) ActionResult<Void> pause(@PathKeysParam PathKeys pathKeys,
      @ActionParam("force") @Optional("false") boolean force); @Action(name = "movePartitions", resourceLevel = ResourceLevel.ENTITY) ActionResult<Void> movePartitions(@PathKeysParam PathKeys pathKeys,
      @ActionParam("partitions") String partitions,
      @ActionParam("targetHost") String targetHost,
      @ActionParam("notify") @Optional("true") boolean notify); @Action(name = "stop", resourceLevel = ResourceLevel.ENTITY) ActionResult<Void> stop(@PathKeysParam PathKeys pathKeys,
      @ActionParam("force") @Optional("false") boolean force); @Action(name = "resume", resourceLevel = ResourceLevel.ENTITY) ActionResult<Void> resume(@PathKeysParam PathKeys pathKeys,
      @ActionParam("force") @Optional("false") boolean force); @Action(name = "pauseSourcePartitions", resourceLevel = ResourceLevel.ENTITY) ActionResult<Void> pauseSourcePartitions(@PathKeysParam PathKeys pathKeys,
      @ActionParam("sourcePartitions") StringMap sourcePartitions); @Action(name = "resumeSourcePartitions", resourceLevel = ResourceLevel.ENTITY) ActionResult<Void> resumeSourcePartitions(@PathKeysParam PathKeys pathKeys,
      @ActionParam("sourcePartitions") StringMap sourcePartitions); @Override UpdateResponse delete(String datastreamName); @Override Datastream get(String name); @SuppressWarnings("deprecated") @Override List<Datastream> getAll(@Context PagingContext pagingContext); @SuppressWarnings("deprecated") @Finder("findGroup") List<Datastream> findGroup(@Context PagingContext pagingContext,
      @QueryParam("datastreamName") String datastreamName); @Override CreateResponse create(Datastream datastream); static List<BrooklinMetricInfo> getMetricInfos(); static final String KEY_NAME; }
@Test(enabled = false) public void testCreateGetAllDatastreamsPaging() throws Exception { DatastreamResources resource = new DatastreamResources(_datastreamKafkaCluster.getPrimaryDatastreamServer()); Assert.assertEquals(resource.getAll(NO_PAGING).size(), 0); List<Datastream> datastreams = createDatastreams(resource, "TestDatastream-", 10); int skip = 2; int limit = 5; List<Datastream> queryStreams = resource.getAll(new PagingContext(skip, limit)); queryStreams.forEach(queryStream -> Assert.assertNotNull(queryStream.getDestination())); Assert.assertEquals( datastreams.stream().map(Datastream::getName).skip(skip).limit(limit).collect(Collectors.toSet()), queryStreams.stream().map(Datastream::getName).collect(Collectors.toSet())); }