@Test(expectedExceptions = RetriesExhaustedException.class) @SuppressWarnings("unchecked") public void testGetDatastreamRetryExhaust() throws Exception { RestClient httpRestClient = mock(RestClient.class); ResponseFuture<IdResponse<String>> response = mock(ResponseFuture.class); when(httpRestClient.sendRequest(any(Request.class))).thenReturn(response); when(response.getResponse()).thenThrow(new RemoteInvocationException(new TimeoutException())); Properties restClientConfig = new Properties(); restClientConfig.put(DatastreamRestClient.CONFIG_RETRY_PERIOD_MS, "10"); restClientConfig.put(DatastreamRestClient.CONFIG_RETRY_TIMEOUT_MS, "200"); DatastreamRestClient restClient = new DatastreamRestClient(httpRestClient, restClientConfig); restClient.getDatastream("datastreamName"); }
public Datastream getDatastream(String datastreamName) { Instant startTime = Instant.now(); Datastream datastream = PollUtils.poll(() -> { try { return doGetDatastream(datastreamName); } catch (RemoteInvocationException e) { if (ExceptionUtils.getRootCause(e) instanceof TimeoutException) { LOG.warn("Timeout: getDatastream. May retry...", e); return null; } if (isNotFoundHttpStatus(e)) { LOG.warn(String.format("Datastream {%s} is not found", datastreamName), e); throw new DatastreamNotFoundException(datastreamName, e); } else { String errorMessage = String.format("Get Datastream {%s} failed with error.", datastreamName); ErrorLogger.logAndThrowDatastreamRuntimeException(LOG, errorMessage, e); return null; } } }, Objects::nonNull, getRetryPeriodMs(), getRetryTimeoutMs()).orElseThrow(RetriesExhaustedException::new); LOG.info("getDatastream for datastream {} took {} ms", datastreamName, Duration.between(startTime, Instant.now()).toMillis()); return datastream; }
DatastreamRestClient { public Datastream getDatastream(String datastreamName) { Instant startTime = Instant.now(); Datastream datastream = PollUtils.poll(() -> { try { return doGetDatastream(datastreamName); } catch (RemoteInvocationException e) { if (ExceptionUtils.getRootCause(e) instanceof TimeoutException) { LOG.warn("Timeout: getDatastream. May retry...", e); return null; } if (isNotFoundHttpStatus(e)) { LOG.warn(String.format("Datastream {%s} is not found", datastreamName), e); throw new DatastreamNotFoundException(datastreamName, e); } else { String errorMessage = String.format("Get Datastream {%s} failed with error.", datastreamName); ErrorLogger.logAndThrowDatastreamRuntimeException(LOG, errorMessage, e); return null; } } }, Objects::nonNull, getRetryPeriodMs(), getRetryTimeoutMs()).orElseThrow(RetriesExhaustedException::new); LOG.info("getDatastream for datastream {} took {} ms", datastreamName, Duration.between(startTime, Instant.now()).toMillis()); return datastream; } }
DatastreamRestClient { public Datastream getDatastream(String datastreamName) { Instant startTime = Instant.now(); Datastream datastream = PollUtils.poll(() -> { try { return doGetDatastream(datastreamName); } catch (RemoteInvocationException e) { if (ExceptionUtils.getRootCause(e) instanceof TimeoutException) { LOG.warn("Timeout: getDatastream. May retry...", e); return null; } if (isNotFoundHttpStatus(e)) { LOG.warn(String.format("Datastream {%s} is not found", datastreamName), e); throw new DatastreamNotFoundException(datastreamName, e); } else { String errorMessage = String.format("Get Datastream {%s} failed with error.", datastreamName); ErrorLogger.logAndThrowDatastreamRuntimeException(LOG, errorMessage, e); return null; } } }, Objects::nonNull, getRetryPeriodMs(), getRetryTimeoutMs()).orElseThrow(RetriesExhaustedException::new); LOG.info("getDatastream for datastream {} took {} ms", datastreamName, Duration.between(startTime, Instant.now()).toMillis()); return datastream; } @Deprecated  DatastreamRestClient(RestClient restClient);  DatastreamRestClient(RestClient restClient, Properties config); }
DatastreamRestClient { public Datastream getDatastream(String datastreamName) { Instant startTime = Instant.now(); Datastream datastream = PollUtils.poll(() -> { try { return doGetDatastream(datastreamName); } catch (RemoteInvocationException e) { if (ExceptionUtils.getRootCause(e) instanceof TimeoutException) { LOG.warn("Timeout: getDatastream. May retry...", e); return null; } if (isNotFoundHttpStatus(e)) { LOG.warn(String.format("Datastream {%s} is not found", datastreamName), e); throw new DatastreamNotFoundException(datastreamName, e); } else { String errorMessage = String.format("Get Datastream {%s} failed with error.", datastreamName); ErrorLogger.logAndThrowDatastreamRuntimeException(LOG, errorMessage, e); return null; } } }, Objects::nonNull, getRetryPeriodMs(), getRetryTimeoutMs()).orElseThrow(RetriesExhaustedException::new); LOG.info("getDatastream for datastream {} took {} ms", datastreamName, Duration.between(startTime, Instant.now()).toMillis()); return datastream; } @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); }
DatastreamRestClient { public Datastream getDatastream(String datastreamName) { Instant startTime = Instant.now(); Datastream datastream = PollUtils.poll(() -> { try { return doGetDatastream(datastreamName); } catch (RemoteInvocationException e) { if (ExceptionUtils.getRootCause(e) instanceof TimeoutException) { LOG.warn("Timeout: getDatastream. May retry...", e); return null; } if (isNotFoundHttpStatus(e)) { LOG.warn(String.format("Datastream {%s} is not found", datastreamName), e); throw new DatastreamNotFoundException(datastreamName, e); } else { String errorMessage = String.format("Get Datastream {%s} failed with error.", datastreamName); ErrorLogger.logAndThrowDatastreamRuntimeException(LOG, errorMessage, e); return null; } } }, Objects::nonNull, getRetryPeriodMs(), getRetryTimeoutMs()).orElseThrow(RetriesExhaustedException::new); LOG.info("getDatastream for datastream {} took {} ms", datastreamName, Duration.between(startTime, Instant.now()).toMillis()); return datastream; } @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; }