JobNodeConfigurationGenerator { JobConfig generateJobConfig(JobNode jobNode, String executionPlanJson) { if (jobNode.isLegacyTaskApplication()) { return new JobConfig(jobNode.getConfig()); } Map<String, String> generatedConfig = new HashMap<>(); generatedConfig.put(JobConfig.JOB_NAME, jobNode.getJobName()); generatedConfig.put(JobConfig.JOB_ID, jobNode.getJobId()); Map<String, StreamEdge> inEdges = jobNode.getInEdges(); Map<String, StreamEdge> outEdges = jobNode.getOutEdges(); Collection<OperatorSpec> reachableOperators = jobNode.getReachableOperators(); List<StoreDescriptor> stores = getStoreDescriptors(reachableOperators); Map<String, TableDescriptor> reachableTables = getReachableTables(reachableOperators, jobNode); Config originalConfig = jobNode.getConfig(); final Set<String> inputs = new HashSet<>(); final Set<String> broadcastInputs = new HashSet<>(); for (StreamEdge inEdge : inEdges.values()) { String formattedSystemStream = inEdge.getName(); if (inEdge.isBroadcast()) { if (inEdge.getPartitionCount() > 1) { broadcastInputs.add(formattedSystemStream + "#[0-" + (inEdge.getPartitionCount() - 1) + "]"); } else { broadcastInputs.add(formattedSystemStream + "#0"); } } else { inputs.add(formattedSystemStream); } } configureBroadcastInputs(generatedConfig, originalConfig, broadcastInputs); configureWindowInterval(generatedConfig, originalConfig, reachableOperators); stores.forEach(sd -> generatedConfig.putAll(sd.getStorageConfigs())); generatedConfig.put(CONFIG_INTERNAL_EXECUTION_PLAN, executionPlanJson); inEdges.values().stream().filter(StreamEdge::isIntermediate) .forEach(intermediateEdge -> generatedConfig.putAll(intermediateEdge.generateConfig())); configureSerdes(generatedConfig, inEdges, outEdges, stores, reachableTables.keySet(), jobNode); configureTables(generatedConfig, originalConfig, reachableTables, inputs); generatedConfig.put(TaskConfig.INPUT_STREAMS, Joiner.on(',').join(inputs)); LOG.info("Job {} has generated configs {}", jobNode.getJobNameAndId(), generatedConfig); return new JobConfig(mergeConfig(originalConfig, generatedConfig)); }    }
@Test public void testConfigureSerdesWithRepartitionJoinApplication() { mockStreamAppDesc = new StreamApplicationDescriptorImpl(getRepartitionJoinStreamApplication(), mockConfig); configureJobNode(mockStreamAppDesc); JobNodeConfigurationGenerator configureGenerator = new JobNodeConfigurationGenerator(); JobConfig jobConfig = configureGenerator.generateJobConfig(mockJobNode, "testJobGraphJson"); Config expectedJobConfig = getExpectedJobConfig(mockConfig, mockJobNode.getInEdges()); validateJobConfig(expectedJobConfig, jobConfig); assertEquals("3600000", jobConfig.get(TaskConfig.WINDOW_MS)); Map<String, Serde> deserializedSerdes = validateAndGetDeserializedSerdes(jobConfig, 5); validateStreamConfigures(jobConfig, deserializedSerdes); validateJoinStoreConfigures(jobConfig, deserializedSerdes); }
@Test public void testConfigureSerdesForRepartitionWithNoDefaultSystem() { mockStreamAppDesc = new StreamApplicationDescriptorImpl(getRepartitionOnlyStreamApplication(), mockConfig); configureJobNode(mockStreamAppDesc); JobNodeConfigurationGenerator configureGenerator = new JobNodeConfigurationGenerator(); JobConfig jobConfig = configureGenerator.generateJobConfig(mockJobNode, "testJobGraphJson"); Config expectedJobConfig = getExpectedJobConfig(mockConfig, mockJobNode.getInEdges()); validateJobConfig(expectedJobConfig, jobConfig); Map<String, Serde> deserializedSerdes = validateAndGetDeserializedSerdes(jobConfig, 2); validateStreamConfigures(jobConfig, null); String partitionByKeySerde = jobConfig.get("streams.jobName-jobId-partition_by-p1.samza.key.serde"); String partitionByMsgSerde = jobConfig.get("streams.jobName-jobId-partition_by-p1.samza.msg.serde"); assertTrue("Serialized serdes should not contain intermediate stream key serde", !deserializedSerdes.containsKey(partitionByKeySerde)); assertTrue("Serialized serdes should not contain intermediate stream msg serde", !deserializedSerdes.containsKey(partitionByMsgSerde)); }
@Test public void testGenerateJobConfigWithTaskApplication() { TaskApplicationDescriptorImpl taskAppDesc = new TaskApplicationDescriptorImpl(getTaskApplication(), mockConfig); configureJobNode(taskAppDesc); JobNodeConfigurationGenerator configureGenerator = new JobNodeConfigurationGenerator(); JobConfig jobConfig = configureGenerator.generateJobConfig(mockJobNode, "testJobGraphJson"); Config expectedJobConfig = getExpectedJobConfig(mockConfig, mockJobNode.getInEdges()); validateJobConfig(expectedJobConfig, jobConfig); Map<String, Serde> deserializedSerdes = validateAndGetDeserializedSerdes(jobConfig, 2); validateStreamConfigures(jobConfig, deserializedSerdes); }
@Test public void testGenerateJobConfigWithLegacyTaskApplication() { TaskApplicationDescriptorImpl taskAppDesc = new TaskApplicationDescriptorImpl(getLegacyTaskApplication(), mockConfig); configureJobNode(taskAppDesc); Map<String, String> originConfig = new HashMap<>(mockConfig); JobNodeConfigurationGenerator configureGenerator = new JobNodeConfigurationGenerator(); JobConfig jobConfig = configureGenerator.generateJobConfig(mockJobNode, ""); Map<String, String> generatedConfig = new HashMap<>(jobConfig); assertEquals(originConfig, generatedConfig); }
@Test public void testBroadcastStreamApplication() { mockStreamAppDesc = new StreamApplicationDescriptorImpl(getBroadcastOnlyStreamApplication(defaultSerde), mockConfig); configureJobNode(mockStreamAppDesc); JobNodeConfigurationGenerator configureGenerator = new JobNodeConfigurationGenerator(); JobConfig jobConfig = configureGenerator.generateJobConfig(mockJobNode, "testJobGraphJson"); Config expectedJobConfig = getExpectedJobConfig(mockConfig, mockJobNode.getInEdges()); validateJobConfig(expectedJobConfig, jobConfig); Map<String, Serde> deserializedSerdes = validateAndGetDeserializedSerdes(jobConfig, 2); validateStreamSerdeConfigure(broadcastInputDesriptor.getStreamId(), jobConfig, deserializedSerdes); validateIntermediateStreamConfigure(broadcastInputDesriptor.getStreamId(), broadcastInputDesriptor.getPhysicalName().get(), jobConfig); }
@Test public void testStreamApplicationWithTableAndSideInput() { mockStreamAppDesc = new StreamApplicationDescriptorImpl(getRepartitionJoinStreamApplication(), mockConfig); GenericInputDescriptor<KV<String, Object>> sideInput1 = inputSystemDescriptor.getInputDescriptor("sideInput1", defaultSerde); BaseTableDescriptor mockTableDescriptor = new MockLocalTableDescriptor("testTable", defaultSerde) .withSideInputs(Arrays.asList(sideInput1.getStreamId())) .withSideInputsProcessor(mock(SideInputsProcessor.class, withSettings().serializable())) .withConfig("mock.table.provider.config", "mock.config.value"); mockStreamAppDesc.getInputStream(sideInput1).sendTo(mockStreamAppDesc.getTable(mockTableDescriptor)); StreamEdge sideInputEdge = new StreamEdge(new StreamSpec(sideInput1.getStreamId(), "sideInput1", inputSystemDescriptor.getSystemName()), false, false, mockConfig); Map<String, String> configs = new HashMap<>(mockConfig); configs.putAll(sideInputEdge.generateConfig()); mockConfig = spy(new MapConfig(configs)); configureJobNode(mockStreamAppDesc); JobNodeConfigurationGenerator configureGenerator = new JobNodeConfigurationGenerator(); JobConfig jobConfig = configureGenerator.generateJobConfig(mockJobNode, "testJobGraphJson"); Config expectedJobConfig = getExpectedJobConfig(mockConfig, mockJobNode.getInEdges()); validateJobConfig(expectedJobConfig, jobConfig); Map<String, Serde> deserializedSerdes = validateAndGetDeserializedSerdes(jobConfig, 5); validateTableConfigure(jobConfig, deserializedSerdes, mockTableDescriptor); }
@Test public void testTaskApplicationWithTableAndSideInput() { GenericInputDescriptor<KV<String, Object>> sideInput1 = inputSystemDescriptor.getInputDescriptor("sideInput1", defaultSerde); BaseTableDescriptor mockTableDescriptor = new MockLocalTableDescriptor("testTable", defaultSerde) .withSideInputs(Arrays.asList(sideInput1.getStreamId())) .withSideInputsProcessor(mock(SideInputsProcessor.class, withSettings().serializable())) .withConfig("mock.table.provider.config", "mock.config.value"); StreamEdge sideInputEdge = new StreamEdge(new StreamSpec(sideInput1.getStreamId(), "sideInput1", inputSystemDescriptor.getSystemName()), false, false, mockConfig); Map<String, String> configs = new HashMap<>(mockConfig); configs.putAll(sideInputEdge.generateConfig()); mockConfig = spy(new MapConfig(configs)); TaskApplicationDescriptorImpl taskAppDesc = new TaskApplicationDescriptorImpl(getTaskApplication(), mockConfig); taskAppDesc.withTable(mockTableDescriptor); taskAppDesc.withInputStream(inputSystemDescriptor.getInputDescriptor("sideInput1", defaultSerde)); configureJobNode(taskAppDesc); JobNodeConfigurationGenerator configureGenerator = new JobNodeConfigurationGenerator(); JobConfig jobConfig = configureGenerator.generateJobConfig(mockJobNode, "testJobGraphJson"); Config expectedJobConfig = getExpectedJobConfig(mockConfig, mockJobNode.getInEdges()); validateJobConfig(expectedJobConfig, jobConfig); Map<String, Serde> deserializedSerdes = validateAndGetDeserializedSerdes(jobConfig, 2); validateStreamConfigures(jobConfig, deserializedSerdes); validateTableConfigure(jobConfig, deserializedSerdes, mockTableDescriptor); }
@Test public void testConfigRewriter() { Map<String, String> configs = new HashMap<>(mockConfig); String streamCfgToOverride = String.format("streams.%s.samza.system", intermediateInputDescriptor.getStreamId()); configs.put(String.format(JobConfig.CONFIG_REWRITER_CLASS, "mock"), MockConfigRewriter.class.getName()); configs.put(JobConfig.CONFIG_REWRITERS, "mock"); configs.put(String.format("job.config.rewriter.mock.%s", streamCfgToOverride), "rewritten-system"); mockConfig = spy(new MapConfig(configs)); mockStreamAppDesc = new StreamApplicationDescriptorImpl(getRepartitionJoinStreamApplication(), mockConfig); configureJobNode(mockStreamAppDesc); JobNodeConfigurationGenerator configureGenerator = new JobNodeConfigurationGenerator(); JobConfig jobConfig = configureGenerator.generateJobConfig(mockJobNode, "testJobGraphJson"); Config expectedConfig = getExpectedJobConfig(mockConfig, mockJobNode.getInEdges()); validateJobConfig(expectedConfig, jobConfig); assertEquals("rewritten-system", jobConfig.get(streamCfgToOverride)); }