DatastreamResources extends CollectionResourceTemplate<String, Datastream> { @Override public CreateResponse create(Datastream datastream) { try { LOG.info("Create datastream called with datastream {}", datastream); if (LOG.isDebugEnabled()) { LOG.debug("Handling request on object: {} thread: {}", this, Thread.currentThread()); } _dynamicMetricsManager.createOrUpdateMeter(CLASS_NAME, CREATE_CALL, 1); DatastreamUtils.validateNewDatastream(datastream); Validate.isTrue(datastream.hasName(), "Must specify name of Datastream!"); Validate.isTrue(datastream.hasConnectorName(), "Must specify connectorType!"); Validate.isTrue(datastream.hasSource(), "Must specify source of Datastream!"); Validate.isTrue(datastream.hasMetadata(), "Missing metadata for Datastream!"); StringMap metadataMap = datastream.getMetadata(); Validate.isTrue(metadataMap.containsKey(DatastreamMetadataConstants.OWNER_KEY), "Must specify owner of Datastream"); if (datastream.hasDestination() && datastream.getDestination().hasConnectionString()) { metadataMap.put(DatastreamMetadataConstants.IS_USER_MANAGED_DESTINATION_KEY, "true"); } String datastreamName = datastream.getName(); String trimmedDatastreamName = datastreamName.trim(); if (!Objects.equals(datastreamName, trimmedDatastreamName)) { datastream.setName(trimmedDatastreamName); LOG.info("Leading and/or trailing whitespace found in datastream name. Name trimmed from '{}' to '{}'", datastreamName, trimmedDatastreamName); } Instant startTime = Instant.now(); LOG.debug("Sanity check is finished, initializing datastream"); if (_store.getDatastream(datastream.getName()) != null) { throw new DatastreamAlreadyExistsException(); } _coordinator.initializeDatastream(datastream); LOG.debug("Persisting initialized datastream to zookeeper: {}", datastream); _store.createDatastream(datastream.getName(), datastream); Duration delta = Duration.between(startTime, Instant.now()); CREATE_CALL_LATENCY_MS.set(delta.toMillis()); LOG.info("Datastream persisted to zookeeper, total time used: {} ms", delta.toMillis()); return new CreateResponse(datastream.getName(), HttpStatus.S_201_CREATED); } catch (IllegalArgumentException e) { _dynamicMetricsManager.createOrUpdateMeter(CLASS_NAME, CALL_ERROR, 1); _errorLogger.logAndThrowRestLiServiceException(HttpStatus.S_400_BAD_REQUEST, "Invalid input params for create request", e); } catch (DatastreamValidationException e) { _dynamicMetricsManager.createOrUpdateMeter(CLASS_NAME, CALL_ERROR, 1); _errorLogger.logAndThrowRestLiServiceException(HttpStatus.S_400_BAD_REQUEST, "Failed to initialize Datastream: " + datastream, e); } catch (DatastreamAlreadyExistsException e) { _dynamicMetricsManager.createOrUpdateMeter(CLASS_NAME, CALL_ERROR, 1); _errorLogger.logAndThrowRestLiServiceException(HttpStatus.S_409_CONFLICT, "Datastream with the same name already exists: " + datastream, e); } catch (AuthorizationException e) { _dynamicMetricsManager.createOrUpdateMeter(CLASS_NAME, CALL_ERROR, 1); _errorLogger.logAndThrowRestLiServiceException(HttpStatus.S_401_UNAUTHORIZED, "Datastream creation denied due to insufficient authorization: " + datastream, e); } catch (Exception e) { _dynamicMetricsManager.createOrUpdateMeter(CLASS_NAME, CALL_ERROR, 1); _errorLogger.logAndThrowRestLiServiceException(HttpStatus.S_500_INTERNAL_SERVER_ERROR, "Unexpected error during datastream creation: " + datastream, e); } return null; } 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 public void testCreateEncryptedDatastream() throws Exception { DatastreamResources resource = new DatastreamResources(_datastreamKafkaCluster.getPrimaryDatastreamServer()); Datastream encryptedDS = generateEncryptedDatastream(1, true, true); CreateResponse response = resource.create(encryptedDS); Assert.assertNull(response.getError()); Assert.assertEquals(response.getStatus(), HttpStatus.S_201_CREATED); Datastream justByotDS = generateEncryptedDatastream(3, false, true); DatastreamDestination destination = new DatastreamDestination() .setConnectionString("http: justByotDS.setDestination(destination); response = resource.create(justByotDS); Assert.assertNull(response.getError()); Assert.assertEquals(response.getStatus(), HttpStatus.S_201_CREATED); }
@Test public void testCreateDatastream() throws Exception { DatastreamResources resource = new DatastreamResources(_datastreamKafkaCluster.getPrimaryDatastreamServer()); Set<String> missingFields = new HashSet<>(); Datastream fullDatastream = generateDatastream(0); CreateResponse response = resource.create(fullDatastream); Assert.assertNull(response.getError()); Assert.assertEquals(response.getStatus(), HttpStatus.S_201_CREATED); Datastream whitespaceDatastream = generateDatastream(1); String originalName = whitespaceDatastream.getName(); Assert.assertEquals(originalName, originalName.trim()); whitespaceDatastream.setName(String.format(" %s ", whitespaceDatastream.getName())); response = resource.create(whitespaceDatastream); Assert.assertNull(response.getError()); Assert.assertEquals(response.getStatus(), HttpStatus.S_201_CREATED); Assert.assertEquals(response.getId(), originalName); missingFields.add("target"); Datastream allRequiredFields = generateDatastream(2, missingFields); response = resource.create(allRequiredFields); Assert.assertNull(response.getError()); Assert.assertEquals(response.getStatus(), HttpStatus.S_201_CREATED); missingFields.clear(); missingFields.add("name"); Datastream noName = generateDatastream(3, missingFields); checkBadRequest(() -> resource.create(noName)); missingFields.clear(); missingFields.add("connectorType"); Datastream noConnectorType = generateDatastream(4, missingFields); checkBadRequest(() -> resource.create(noConnectorType)); missingFields.clear(); missingFields.add("source"); Datastream noSource = generateDatastream(5, missingFields); checkBadRequest(() -> resource.create(noSource)); missingFields.clear(); missingFields.add("metadata"); Datastream noMetadata = generateDatastream(6, missingFields); checkBadRequest(() -> resource.create(noMetadata)); Datastream noOwner = generateDatastream(6); noOwner.getMetadata().remove("owner"); checkBadRequest(() -> resource.create(noOwner)); Datastream badConnector = generateDatastream(6); badConnector.setConnectorName("BadConnector"); checkBadRequest(() -> resource.create(badConnector)); Datastream badSource = generateDatastream(6); badSource.getSource().setConnectionString("BadSource"); checkBadRequest(() -> resource.create(badSource)); checkBadRequest(() -> resource.create(allRequiredFields), HttpStatus.S_409_CONFLICT); Datastream undefinedProvider = generateDatastream(7); String undefinedProviderName = "whatsoever"; undefinedProvider.setTransportProviderName(undefinedProviderName); try { resource.create(undefinedProvider); Assert.fail("Should have failed for undefined provider name"); } catch (RestLiServiceException e) { Assert.assertNotNull(e.getMessage()); Assert.assertTrue(e.getMessage().contains(undefinedProviderName)); } }