DatastreamRestClient { public void deleteDatastream(String datastreamName) { Instant startTime = Instant.now(); PollUtils.poll(() -> { DeleteRequest<Datastream> request = _builders.delete().id(datastreamName).build(); ResponseFuture<EmptyRecord> response = _restClient.sendRequest(request); try { return response.getResponse(); } catch (RemoteInvocationException e) { if (ExceptionUtils.getRootCause(e) instanceof TimeoutException) { LOG.warn("Timeout: deleteDatastream. May retry...", e); return null; } String errorMessage = String.format("Delete Datastream %s failed with error.", datastreamName); if (isNotFoundHttpStatus(e)) { LOG.error(errorMessage, e); throw new DatastreamNotFoundException(datastreamName, e); } else { ErrorLogger.logAndThrowDatastreamRuntimeException(LOG, errorMessage, e); } return null; } }, Objects::nonNull, getRetryPeriodMs(), getRetryTimeoutMs()).orElseThrow(RetriesExhaustedException::new); LOG.info("deleteDatastream for datastream {} took {} ms", datastreamName, Duration.between(startTime, Instant.now()).toMillis()); } @Deprecated  DatastreamRestClient(RestClient restClient);  DatastreamRestClient(RestClient restClient, Properties config); Datastream getDatastream(String datastreamName); Datastream waitTillDatastreamIsInitialized(String datastreamName, long timeoutMs); void waitTillDatastreamIsDeleted(String datastreamName, long timeoutMs); List<Datastream> getAllDatastreams(); List<Datastream> getAllDatastreams(int start, int count); void createDatastream(Datastream datastream); void updateDatastream(Datastream datastream); void updateDatastream(List<Datastream> datastreams); void deleteDatastream(String datastreamName); boolean datastreamExists(String datastreamName); void pause(String datastreamName); void pause(String datastreamName, boolean force); void stop(String datastreamName); void resume(String datastreamName); void resume(String datastreamName, boolean force); void movePartitions(String datastreamName, String partitions, String targetHost); List<Datastream> findGroup(String datastreamName); void pauseSourcePartitions(String datastreamName, StringMap sourcePartitions); void resumeSourcePartitions(String datastreamName, StringMap sourcePartitions); static final String CONFIG_RETRY_PERIOD_MS; static final String CONFIG_RETRY_TIMEOUT_MS; }
@Test(expectedExceptions = DatastreamNotFoundException.class) public void testDeleteDatastream() throws Exception { Datastream datastream = generateDatastream(2); LOG.info("Datastream : {}", datastream); DatastreamRestClient restClient = createRestClient(); restClient.createDatastream(datastream); restClient.waitTillDatastreamIsInitialized(datastream.getName(), Duration.ofMinutes(2).toMillis()); restClient.deleteDatastream(datastream.getName()); restClient.waitTillDatastreamIsDeleted(datastream.getName(), Duration.ofMinutes(2).toMillis()); restClient.getDatastream(datastream.getName()); }
@Test(expectedExceptions = DatastreamNotFoundException.class) public void testDeleteNonExistentDatastream() throws Exception { DatastreamRestClient restClient = createRestClient(); restClient.deleteDatastream("NoSuchDatastream"); }