@Test public void testBadCreateThrowsWithServerErrorRetained() throws Exception { try { DatastreamRestClient restClient = createRestClient(); restClient.createDatastream(new Datastream()); } catch (DatastreamRuntimeException e) { RestLiResponseException re = (RestLiResponseException) e.getCause(); Assert.assertNotNull(re.getServiceErrorMessage()); } }
public void createDatastream(Datastream datastream) { Instant startTime = Instant.now(); String creationUid = UUID.randomUUID().toString(); PollUtils.poll(() -> { if (!datastream.hasMetadata()) { datastream.setMetadata(new StringMap()); } datastream.getMetadata().put(DATASTREAM_UUID, creationUid); CreateIdRequest<String, Datastream> request = _builders.create().input(datastream).build(); ResponseFuture<IdResponse<String>> datastreamResponseFuture = _restClient.sendRequest(request); try { return datastreamResponseFuture.getResponse(); } catch (RemoteInvocationException e) { if (ExceptionUtils.getRootCause(e) instanceof TimeoutException) { LOG.warn("Timeout: createDatastream. May retry...", e); return null; } if (e instanceof RestLiResponseException) { int errorCode = ((RestLiResponseException) e).getStatus(); if (errorCode == HttpStatus.S_409_CONFLICT.getCode()) { Datastream existingDatastream = getDatastream(datastream.getName()); Optional<String> existingUid = Optional.ofNullable(existingDatastream.getMetadata().get(DATASTREAM_UUID)); if (existingUid.isPresent() && existingUid.get().equals(creationUid)) { return existingDatastream; } String msg = String.format("Datastream %s exists with the same name. Requested datastream %s", existingDatastream.toString(), datastream.toString()); LOG.warn(msg, e); throw new DatastreamAlreadyExistsException(msg); } else if (errorCode == HttpStatus.S_403_FORBIDDEN.getCode()) { String msg = "Client is not authorized to invoke Datastream-CREATE, stream=" + datastream.getName(); ErrorLogger.logAndThrowDatastreamRuntimeException(LOG, msg, e); } } String errorMessage = String.format("Create Datastream %s failed with error.", datastream); ErrorLogger.logAndThrowDatastreamRuntimeException(LOG, errorMessage, e); return null; } }, Objects::nonNull, getRetryPeriodMs(), getRetryTimeoutMs()).orElseThrow(RetriesExhaustedException::new); LOG.info("createDatastream for datastream {} took {} ms", datastream, Duration.between(startTime, Instant.now()).toMillis()); }
DatastreamRestClient { public void createDatastream(Datastream datastream) { Instant startTime = Instant.now(); String creationUid = UUID.randomUUID().toString(); PollUtils.poll(() -> { if (!datastream.hasMetadata()) { datastream.setMetadata(new StringMap()); } datastream.getMetadata().put(DATASTREAM_UUID, creationUid); CreateIdRequest<String, Datastream> request = _builders.create().input(datastream).build(); ResponseFuture<IdResponse<String>> datastreamResponseFuture = _restClient.sendRequest(request); try { return datastreamResponseFuture.getResponse(); } catch (RemoteInvocationException e) { if (ExceptionUtils.getRootCause(e) instanceof TimeoutException) { LOG.warn("Timeout: createDatastream. May retry...", e); return null; } if (e instanceof RestLiResponseException) { int errorCode = ((RestLiResponseException) e).getStatus(); if (errorCode == HttpStatus.S_409_CONFLICT.getCode()) { Datastream existingDatastream = getDatastream(datastream.getName()); Optional<String> existingUid = Optional.ofNullable(existingDatastream.getMetadata().get(DATASTREAM_UUID)); if (existingUid.isPresent() && existingUid.get().equals(creationUid)) { return existingDatastream; } String msg = String.format("Datastream %s exists with the same name. Requested datastream %s", existingDatastream.toString(), datastream.toString()); LOG.warn(msg, e); throw new DatastreamAlreadyExistsException(msg); } else if (errorCode == HttpStatus.S_403_FORBIDDEN.getCode()) { String msg = "Client is not authorized to invoke Datastream-CREATE, stream=" + datastream.getName(); ErrorLogger.logAndThrowDatastreamRuntimeException(LOG, msg, e); } } String errorMessage = String.format("Create Datastream %s failed with error.", datastream); ErrorLogger.logAndThrowDatastreamRuntimeException(LOG, errorMessage, e); return null; } }, Objects::nonNull, getRetryPeriodMs(), getRetryTimeoutMs()).orElseThrow(RetriesExhaustedException::new); LOG.info("createDatastream for datastream {} took {} ms", datastream, Duration.between(startTime, Instant.now()).toMillis()); } }
DatastreamRestClient { public void createDatastream(Datastream datastream) { Instant startTime = Instant.now(); String creationUid = UUID.randomUUID().toString(); PollUtils.poll(() -> { if (!datastream.hasMetadata()) { datastream.setMetadata(new StringMap()); } datastream.getMetadata().put(DATASTREAM_UUID, creationUid); CreateIdRequest<String, Datastream> request = _builders.create().input(datastream).build(); ResponseFuture<IdResponse<String>> datastreamResponseFuture = _restClient.sendRequest(request); try { return datastreamResponseFuture.getResponse(); } catch (RemoteInvocationException e) { if (ExceptionUtils.getRootCause(e) instanceof TimeoutException) { LOG.warn("Timeout: createDatastream. May retry...", e); return null; } if (e instanceof RestLiResponseException) { int errorCode = ((RestLiResponseException) e).getStatus(); if (errorCode == HttpStatus.S_409_CONFLICT.getCode()) { Datastream existingDatastream = getDatastream(datastream.getName()); Optional<String> existingUid = Optional.ofNullable(existingDatastream.getMetadata().get(DATASTREAM_UUID)); if (existingUid.isPresent() && existingUid.get().equals(creationUid)) { return existingDatastream; } String msg = String.format("Datastream %s exists with the same name. Requested datastream %s", existingDatastream.toString(), datastream.toString()); LOG.warn(msg, e); throw new DatastreamAlreadyExistsException(msg); } else if (errorCode == HttpStatus.S_403_FORBIDDEN.getCode()) { String msg = "Client is not authorized to invoke Datastream-CREATE, stream=" + datastream.getName(); ErrorLogger.logAndThrowDatastreamRuntimeException(LOG, msg, e); } } String errorMessage = String.format("Create Datastream %s failed with error.", datastream); ErrorLogger.logAndThrowDatastreamRuntimeException(LOG, errorMessage, e); return null; } }, Objects::nonNull, getRetryPeriodMs(), getRetryTimeoutMs()).orElseThrow(RetriesExhaustedException::new); LOG.info("createDatastream for datastream {} took {} ms", datastream, Duration.between(startTime, Instant.now()).toMillis()); } @Deprecated  DatastreamRestClient(RestClient restClient);  DatastreamRestClient(RestClient restClient, Properties config); }
DatastreamRestClient { public void createDatastream(Datastream datastream) { Instant startTime = Instant.now(); String creationUid = UUID.randomUUID().toString(); PollUtils.poll(() -> { if (!datastream.hasMetadata()) { datastream.setMetadata(new StringMap()); } datastream.getMetadata().put(DATASTREAM_UUID, creationUid); CreateIdRequest<String, Datastream> request = _builders.create().input(datastream).build(); ResponseFuture<IdResponse<String>> datastreamResponseFuture = _restClient.sendRequest(request); try { return datastreamResponseFuture.getResponse(); } catch (RemoteInvocationException e) { if (ExceptionUtils.getRootCause(e) instanceof TimeoutException) { LOG.warn("Timeout: createDatastream. May retry...", e); return null; } if (e instanceof RestLiResponseException) { int errorCode = ((RestLiResponseException) e).getStatus(); if (errorCode == HttpStatus.S_409_CONFLICT.getCode()) { Datastream existingDatastream = getDatastream(datastream.getName()); Optional<String> existingUid = Optional.ofNullable(existingDatastream.getMetadata().get(DATASTREAM_UUID)); if (existingUid.isPresent() && existingUid.get().equals(creationUid)) { return existingDatastream; } String msg = String.format("Datastream %s exists with the same name. Requested datastream %s", existingDatastream.toString(), datastream.toString()); LOG.warn(msg, e); throw new DatastreamAlreadyExistsException(msg); } else if (errorCode == HttpStatus.S_403_FORBIDDEN.getCode()) { String msg = "Client is not authorized to invoke Datastream-CREATE, stream=" + datastream.getName(); ErrorLogger.logAndThrowDatastreamRuntimeException(LOG, msg, e); } } String errorMessage = String.format("Create Datastream %s failed with error.", datastream); ErrorLogger.logAndThrowDatastreamRuntimeException(LOG, errorMessage, e); return null; } }, Objects::nonNull, getRetryPeriodMs(), getRetryTimeoutMs()).orElseThrow(RetriesExhaustedException::new); LOG.info("createDatastream for datastream {} took {} ms", datastream, 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); }
DatastreamRestClient { public void createDatastream(Datastream datastream) { Instant startTime = Instant.now(); String creationUid = UUID.randomUUID().toString(); PollUtils.poll(() -> { if (!datastream.hasMetadata()) { datastream.setMetadata(new StringMap()); } datastream.getMetadata().put(DATASTREAM_UUID, creationUid); CreateIdRequest<String, Datastream> request = _builders.create().input(datastream).build(); ResponseFuture<IdResponse<String>> datastreamResponseFuture = _restClient.sendRequest(request); try { return datastreamResponseFuture.getResponse(); } catch (RemoteInvocationException e) { if (ExceptionUtils.getRootCause(e) instanceof TimeoutException) { LOG.warn("Timeout: createDatastream. May retry...", e); return null; } if (e instanceof RestLiResponseException) { int errorCode = ((RestLiResponseException) e).getStatus(); if (errorCode == HttpStatus.S_409_CONFLICT.getCode()) { Datastream existingDatastream = getDatastream(datastream.getName()); Optional<String> existingUid = Optional.ofNullable(existingDatastream.getMetadata().get(DATASTREAM_UUID)); if (existingUid.isPresent() && existingUid.get().equals(creationUid)) { return existingDatastream; } String msg = String.format("Datastream %s exists with the same name. Requested datastream %s", existingDatastream.toString(), datastream.toString()); LOG.warn(msg, e); throw new DatastreamAlreadyExistsException(msg); } else if (errorCode == HttpStatus.S_403_FORBIDDEN.getCode()) { String msg = "Client is not authorized to invoke Datastream-CREATE, stream=" + datastream.getName(); ErrorLogger.logAndThrowDatastreamRuntimeException(LOG, msg, e); } } String errorMessage = String.format("Create Datastream %s failed with error.", datastream); ErrorLogger.logAndThrowDatastreamRuntimeException(LOG, errorMessage, e); return null; } }, Objects::nonNull, getRetryPeriodMs(), getRetryTimeoutMs()).orElseThrow(RetriesExhaustedException::new); LOG.info("createDatastream for datastream {} took {} ms", datastream, 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; }