NamespaceCatalogResource { @POST @Path("/namespaces/{id}/mapping/bulk") @Timed public Response setServicesToClusterInNamespace(@PathParam("id") Long namespaceId, List<NamespaceServiceClusterMap> mappings, @Context SecurityContext securityContext) { SecurityUtil.checkRoleOrPermissions(authorizer, securityContext, Roles.ROLE_ENVIRONMENT_SUPER_ADMIN, Namespace.NAMESPACE, namespaceId, WRITE); Namespace namespace = environmentService.getNamespace(namespaceId); if (namespace == null) { throw EntityNotFoundException.byId(namespaceId.toString()); } String streamingEngine = namespace.getStreamingEngine(); String timeSeriesDB = namespace.getTimeSeriesDB(); Collection<NamespaceServiceClusterMap> existing = environmentService.listServiceClusterMapping(namespaceId); Optional<NamespaceServiceClusterMap> existingStreamingEngine = existing.stream() .filter(m -> m.getServiceName().equals(streamingEngine)) .findFirst(); if (existingStreamingEngine.isPresent() && !mappings.contains(existingStreamingEngine.get())) { assertNoTopologyReferringNamespaceIsRunning(namespaceId, WSUtils.getUserFromSecurityContext(securityContext)); } assertServiceIsUnique(mappings, streamingEngine); assertServiceIsUnique(mappings, timeSeriesDB); Collection<NamespaceServiceClusterMap> existingMappings = environmentService.listServiceClusterMapping(namespaceId); if (existingMappings != null) { existingMappings.forEach(m -> environmentService.removeServiceClusterMapping(m.getNamespaceId(), m.getServiceName(), m.getClusterId())); } List<NamespaceServiceClusterMap> newMappings = mappings.stream() .map(environmentService::addOrUpdateServiceClusterMapping).collect(toList()); return WSUtils.respondEntities(newMappings, CREATED); } NamespaceCatalogResource(StreamlineAuthorizer authorizer, StreamCatalogService catalogService,
                                  TopologyActionsService topologyActionsService, EnvironmentService environmentService); @GET @Path("/namespaces") @Timed Response listNamespaces(@Context UriInfo uriInfo,
                                 @Context SecurityContext securityContext); @GET @Path("/namespaces/{id}") @Timed Response getNamespaceById(@PathParam("id") Long namespaceId,
                                   @javax.ws.rs.QueryParam("detail") Boolean detail,
                                   @Context SecurityContext securityContext); @GET @Path("/namespaces/name/{namespaceName}") @Timed Response getNamespaceByName(@PathParam("namespaceName") String namespaceName,
                                     @javax.ws.rs.QueryParam("detail") Boolean detail,
                                     @Context SecurityContext securityContext); @Timed @POST @Path("/namespaces") Response addNamespace(Namespace namespace, @Context SecurityContext securityContext); @DELETE @Path("/namespaces/{id}") @Timed Response removeNamespace(@PathParam("id") Long namespaceId, @Context SecurityContext securityContext); @PUT @Path("/namespaces/{id}") @Timed Response addOrUpdateNamespace(@PathParam("id") Long namespaceId,
      Namespace namespace, @Context SecurityContext securityContext); @GET @Path("/namespaces/{id}/mapping") @Timed Response listServiceToClusterMappingInNamespace(@PathParam("id") Long namespaceId, @Context SecurityContext securityContext); @GET @Path("/namespaces/{id}/mapping/{serviceName}") @Timed Response findServicesToClusterMappingInNamespace(@PathParam("id") Long namespaceId,
                                                          @PathParam("serviceName") String serviceName,
                                                          @Context SecurityContext securityContext); @POST @Path("/namespaces/{id}/mapping/bulk") @Timed Response setServicesToClusterInNamespace(@PathParam("id") Long namespaceId,
                                                  List<NamespaceServiceClusterMap> mappings,
                                                  @Context SecurityContext securityContext); @POST @Path("/namespaces/{id}/mapping") @Timed Response mapServiceToClusterInNamespace(@PathParam("id") Long namespaceId,
                                                 NamespaceServiceClusterMap mapping, @Context SecurityContext securityContext); @DELETE @Path("/namespaces/{id}/mapping/{serviceName}/cluster/{clusterId}") @Timed Response unmapServiceToClusterInNamespace(@PathParam("id") Long namespaceId,
                                                   @PathParam("serviceName") String serviceName,
                                                   @PathParam("clusterId") Long clusterId,
                                                   @Context SecurityContext securityContext); @DELETE @Path("/namespaces/{id}/mapping") @Timed Response unmapAllServicesToClusterInNamespace(@PathParam("id") Long namespaceId,
                                                       @Context SecurityContext securityContext);  }
@Test public void testExcludeStreamingEngineViaSetServicesToClusterInNamespace() throws Exception { Long testNamespaceId = 1L; Namespace testNamespace = createTestNamespace(testNamespaceId, TEST_STREAMING_ENGINE, TEST_TIME_SERIES_DB); Collection<NamespaceServiceClusterMap> existingMappings = createTestMappingsForExisting(testNamespaceId); setupExpectationForSimulatingTopologyIsRunning(testNamespaceId, testNamespace, existingMappings); List<NamespaceServiceClusterMap> mappingsToApply = existingMappings.stream() .filter(m -> !m.getServiceName().equals(TEST_STREAMING_ENGINE)).collect(toList()); try { namespaceCatalogResource.setServicesToClusterInNamespace(testNamespaceId, mappingsToApply, securityContext); Assert.fail("Should throw BadRequestException"); } catch (BadRequestException e) { } new Verifications() {{ environmentService.removeServiceClusterMapping(testNamespaceId, anyString, anyLong); times = 0; }}; }
@Test public void testChangeMappingOfStreamingEngineViaSetServicesToClusterInNamespace() throws Exception { Long testNamespaceId = 1L; Namespace testNamespace = createTestNamespace(testNamespaceId, TEST_STREAMING_ENGINE, TEST_TIME_SERIES_DB); Collection<NamespaceServiceClusterMap> existingMappings = createTestMappingsForExisting(testNamespaceId); setupExpectationForSimulatingTopologyIsRunning(testNamespaceId, testNamespace, existingMappings); List<NamespaceServiceClusterMap> mappingsToApply = existingMappings.stream() .filter(m -> !m.getServiceName().equals(TEST_STREAMING_ENGINE)).collect(toList()); mappingsToApply.add(new NamespaceServiceClusterMap(testNamespaceId, TEST_STREAMING_ENGINE, 2L)); try { namespaceCatalogResource.setServicesToClusterInNamespace(testNamespaceId, mappingsToApply, securityContext); Assert.fail("Should throw BadRequestException"); } catch (BadRequestException e) { } new Verifications() {{ environmentService.removeServiceClusterMapping(testNamespaceId, anyString, anyLong); times = 0; }}; }
@Test public void testOverwriteExistingStreamingEngineMappingViaSetServicesToClusterInNamespace() throws Exception { Long testNamespaceId = 1L; Namespace testNamespace = createTestNamespace(testNamespaceId, TEST_STREAMING_ENGINE, TEST_TIME_SERIES_DB); Collection<NamespaceServiceClusterMap> existingMappings = createTestMappingsForExisting(testNamespaceId); new Expectations() {{ environmentService.getNamespace(testNamespaceId); result = testNamespace; environmentService.listServiceClusterMapping(testNamespaceId); result = existingMappings; }}; namespaceCatalogResource.setServicesToClusterInNamespace(testNamespaceId, new ArrayList<>(existingMappings), securityContext); new Verifications() {{ catalogService.listTopologies(); times = 0; topologyActionsService.getRuntimeTopologyId(withAny(new Topology()), anyString); times = 0; environmentService.removeServiceClusterMapping(testNamespaceId, anyString, anyLong); times = existingMappings.size(); environmentService.addOrUpdateServiceClusterMapping(withAny(new NamespaceServiceClusterMap())); times = existingMappings.size(); }}; }
@Test public void testMappingMultipleStreamingEngineViaSetServicesToClusterInNamespace() throws Exception { Long testNamespaceId = 1L; Namespace testNamespace = createTestNamespace(testNamespaceId, TEST_STREAMING_ENGINE, TEST_TIME_SERIES_DB); Collection<NamespaceServiceClusterMap> existingMappings = createTestMappingsForExisting(testNamespaceId); new Expectations() {{ environmentService.getNamespace(testNamespaceId); result = testNamespace; environmentService.listServiceClusterMapping(testNamespaceId); result = existingMappings; }}; List<NamespaceServiceClusterMap> mappingsToApply = Lists.newArrayList( new NamespaceServiceClusterMap(testNamespaceId, TEST_STREAMING_ENGINE, 1L), new NamespaceServiceClusterMap(testNamespaceId, TEST_STREAMING_ENGINE, 2L), new NamespaceServiceClusterMap(testNamespaceId, TEST_TIME_SERIES_DB, 1L), new NamespaceServiceClusterMap(testNamespaceId, "KAFKA", 1L) ); try { namespaceCatalogResource.setServicesToClusterInNamespace(testNamespaceId, mappingsToApply, securityContext); Assert.fail("Should throw BadRequestException"); } catch (BadRequestException e) { } new Verifications() {{ catalogService.listTopologies(); times = 0; topologyActionsService.getRuntimeTopologyId(withAny(new Topology()), anyString); times = 0; environmentService.removeServiceClusterMapping(testNamespaceId, anyString, anyLong); times = 0; }}; }
@Test public void testMappingMultipleTimeSeriesDBViaSetServicesToClusterInNamespace() throws Exception { Long testNamespaceId = 1L; Namespace testNamespace = createTestNamespace(testNamespaceId, TEST_STREAMING_ENGINE, TEST_TIME_SERIES_DB); Collection<NamespaceServiceClusterMap> existingMappings = createTestMappingsForExisting(testNamespaceId); new Expectations() {{ environmentService.getNamespace(testNamespaceId); result = testNamespace; environmentService.listServiceClusterMapping(testNamespaceId); result = existingMappings; }}; List<NamespaceServiceClusterMap> mappingsToApply = Lists.newArrayList( new NamespaceServiceClusterMap(testNamespaceId, TEST_STREAMING_ENGINE, 1L), new NamespaceServiceClusterMap(testNamespaceId, TEST_TIME_SERIES_DB, 1L), new NamespaceServiceClusterMap(testNamespaceId, TEST_TIME_SERIES_DB, 2L), new NamespaceServiceClusterMap(testNamespaceId, "KAFKA", 1L) ); try { namespaceCatalogResource.setServicesToClusterInNamespace(testNamespaceId, mappingsToApply, securityContext); Assert.fail("Should throw BadRequestException"); } catch (BadRequestException e) { } new Verifications() {{ catalogService.listTopologies(); times = 0; topologyActionsService.getRuntimeTopologyId(withAny(new Topology()), anyString); times = 0; environmentService.removeServiceClusterMapping(testNamespaceId, anyString, anyLong); times = 0; }}; }