StreamApplicationDescriptorImpl extends ApplicationDescriptorImpl<StreamApplicationDescriptor> implements StreamApplicationDescriptor { public String getNextOpId(OpCode opCode, String userDefinedId) { if (StringUtils.isNotBlank(userDefinedId) && !ID_PATTERN.matcher(userDefinedId).matches()) { throw new SamzaException("Operator ID must not contain spaces or special characters: " + userDefinedId); } ApplicationConfig applicationConfig = new ApplicationConfig(getConfig()); String nextOpId = String.format("%s-%s-%s-%s", applicationConfig.getAppName(), applicationConfig.getAppId(), opCode.name().toLowerCase(), StringUtils.isNotBlank(userDefinedId) ? userDefinedId.trim() : String.valueOf(nextOpNum)); if (!operatorIds.add(nextOpId)) { throw new SamzaException( String.format("Found duplicate operator ID %s in the graph. Operator IDs must be unique.", nextOpId)); } nextOpNum++; return nextOpId; } StreamApplicationDescriptorImpl(StreamApplication userApp, Config config); @Override MessageStream<M> getInputStream(InputDescriptor<M, ?> inputDescriptor); @Override OutputStream<M> getOutputStream(OutputDescriptor<M, ?> outputDescriptor); @Override Table<KV<K, V>> getTable(TableDescriptor<K, V, ?> tableDescriptor); @Override Set<String> getInputStreamIds(); @Override Set<String> getOutputStreamIds(); @Override Set<String> getIntermediateBroadcastStreamIds(); Map<String, InputOperatorSpec> getInputOperators(); Map<String, OutputStreamImpl> getOutputStreams(); OperatorSpecGraph getOperatorSpecGraph(); String getNextOpId(OpCode opCode, String userDefinedId); String getNextOpId(OpCode opCode); @VisibleForTesting IntermediateMessageStreamImpl<M> getIntermediateStream(String streamId, Serde<M> serde, boolean isBroadcast);  }
@Test public void testGetNextOpIdIncrementsId() { HashMap<String, String> configMap = new HashMap<>(); configMap.put(ApplicationConfig.APP_NAME, "appName"); configMap.put(ApplicationConfig.APP_ID, "1234"); Config config = new MapConfig(configMap); StreamApplicationDescriptorImpl streamAppDesc = new StreamApplicationDescriptorImpl(appDesc -> { }, config); assertEquals("appName-1234-merge-0", streamAppDesc.getNextOpId(OpCode.MERGE, null)); assertEquals("appName-1234-join-customName", streamAppDesc.getNextOpId(OpCode.JOIN, "customName")); assertEquals("appName-1234-map-2", streamAppDesc.getNextOpId(OpCode.MAP, null)); }
@Test(expected = SamzaException.class) public void testGetNextOpIdRejectsDuplicates() { HashMap<String, String> configMap = new HashMap<>(); configMap.put(ApplicationConfig.APP_NAME, "appName"); configMap.put(ApplicationConfig.APP_ID, "1234"); Config config = new MapConfig(configMap); StreamApplicationDescriptorImpl streamAppDesc = new StreamApplicationDescriptorImpl(appDesc -> { }, config); assertEquals("appName-1234-join-customName", streamAppDesc.getNextOpId(OpCode.JOIN, "customName")); streamAppDesc.getNextOpId(OpCode.JOIN, "customName"); }
@Test public void testOpIdValidation() { HashMap<String, String> configMap = new HashMap<>(); configMap.put(ApplicationConfig.APP_NAME, "appName"); configMap.put(ApplicationConfig.APP_ID, "1234"); Config config = new MapConfig(configMap); StreamApplicationDescriptorImpl streamAppDesc = new StreamApplicationDescriptorImpl(appDesc -> { }, config); try { streamAppDesc.getNextOpId(OpCode.FILTER, null); streamAppDesc.getNextOpId(OpCode.FILTER, ""); streamAppDesc.getNextOpId(OpCode.FILTER, " "); streamAppDesc.getNextOpId(OpCode.FILTER, "\t"); } catch (SamzaException e) { fail("Received an error with a null or empty operator ID instead of defaulting to auto-generated ID."); } List<String> validOpIds = ImmutableList.of("op_id", "op-id", "1000", "op_1", "OP_ID"); for (String validOpId: validOpIds) { try { streamAppDesc.getNextOpId(OpCode.FILTER, validOpId); } catch (Exception e) { fail("Received an exception with a valid operator ID: " + validOpId); } } List<String> invalidOpIds = ImmutableList.of("op id", "op#id"); for (String invalidOpId: invalidOpIds) { try { streamAppDesc.getNextOpId(OpCode.FILTER, invalidOpId); fail("Did not receive an exception with an invalid operator ID: " + invalidOpId); } catch (SamzaException e) { } } }