ExecutionPlanner { public ExecutionPlan plan(ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc) { validateConfig(); JobGraph jobGraph = createJobGraph(appDesc); setInputAndOutputStreamPartitionCount(jobGraph); List<StreamSet> joinedStreamSets = groupJoinedStreams(jobGraph); if (!jobGraph.getIntermediateStreamEdges().isEmpty()) { new IntermediateStreamManager(config).calculatePartitions(jobGraph, joinedStreamSets); } joinedStreamSets.forEach(ExecutionPlanner::validatePartitions); return jobGraph; } ExecutionPlanner(Config config, StreamManager streamManager); ExecutionPlan plan(ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc);  }
@Test public void testCalculateJoinInputPartitions() { ExecutionPlanner planner = new ExecutionPlanner(config, streamManager); StreamApplicationDescriptorImpl graphSpec = createStreamGraphWithStreamStreamJoin(); JobGraph jobGraph = (JobGraph) planner.plan(graphSpec); jobGraph.getIntermediateStreams().forEach(edge -> { assertEquals(64, edge.getPartitionCount()); }); }
@Test public void testCalculateOrderSensitiveJoinInputPartitions() { ExecutionPlanner planner = new ExecutionPlanner(config, streamManager); StreamApplicationDescriptorImpl graphSpec = createStreamGraphWithComplexStreamStreamJoin(); JobGraph jobGraph = (JobGraph) planner.plan(graphSpec); jobGraph.getIntermediateStreams().forEach(edge -> { assertEquals(64, edge.getPartitionCount()); }); }
@Test public void testCalculateIntStreamPartitions() { ExecutionPlanner planner = new ExecutionPlanner(config, streamManager); StreamApplicationDescriptorImpl graphSpec = createSimpleGraph(); JobGraph jobGraph = (JobGraph) planner.plan(graphSpec); jobGraph.getIntermediateStreams().forEach(edge -> { assertEquals(64, edge.getPartitionCount()); }); }
@Test public void testCalculateInStreamPartitionsBehindTables() { ExecutionPlanner planner = new ExecutionPlanner(config, streamManager); StreamApplicationDescriptorImpl graphSpec = createStreamGraphWithStreamTableJoin(); JobGraph jobGraph = (JobGraph) planner.plan(graphSpec); jobGraph.getIntermediateStreams().forEach(edge -> { assertEquals(32, edge.getPartitionCount()); }); }
@Test public void testCalculateInStreamPartitionsBehindTablesWithSideInputs() { ExecutionPlanner planner = new ExecutionPlanner(config, streamManager); StreamApplicationDescriptorImpl graphSpec = createStreamGraphWithStreamTableJoinWithSideInputs(); JobGraph jobGraph = (JobGraph) planner.plan(graphSpec); jobGraph.getIntermediateStreams().forEach(edge -> { assertEquals(64, edge.getPartitionCount()); }); }
@Test public void testHandlesVirtualStreamTableJoinCycles() { ExecutionPlanner planner = new ExecutionPlanner(config, streamManager); StreamApplicationDescriptorImpl graphSpec = createStreamGraphWithStreamTableJoinAndSendToSameTable(); planner.plan(graphSpec); }
@Test public void testDefaultPartitions() { Map<String, String> map = new HashMap<>(config); map.put(JobConfig.JOB_INTERMEDIATE_STREAM_PARTITIONS, String.valueOf(DEFAULT_PARTITIONS)); Config cfg = new MapConfig(map); ExecutionPlanner planner = new ExecutionPlanner(cfg, streamManager); StreamApplicationDescriptorImpl graphSpec = createSimpleGraph(); JobGraph jobGraph = (JobGraph) planner.plan(graphSpec); jobGraph.getIntermediateStreams().forEach(edge -> { assertTrue(edge.getPartitionCount() == DEFAULT_PARTITIONS); }); }
@Test public void testBroadcastConfig() { Map<String, String> map = new HashMap<>(config); map.put(String.format(StreamConfig.BROADCAST_FOR_STREAM_ID, "input1"), "true"); Config cfg = new MapConfig(map); ExecutionPlanner planner = new ExecutionPlanner(cfg, streamManager); StreamApplicationDescriptorImpl graphSpec = createSimpleGraph(); JobGraph jobGraph = (JobGraph) planner.plan(graphSpec); StreamEdge edge = jobGraph.getStreamEdge("input1"); Assert.assertTrue(edge.isBroadcast()); Config jobConfig = jobGraph.getJobConfigs().get(0); Assert.assertEquals("system1.input1#[0-63]", jobConfig.get("task.broadcast.inputs")); }
@Test public void testMaxPartitionLimit() { int partitionLimit = IntermediateStreamManager.MAX_INFERRED_PARTITIONS; ExecutionPlanner planner = new ExecutionPlanner(config, streamManager); StreamApplicationDescriptorImpl graphSpec = new StreamApplicationDescriptorImpl(appDesc -> { MessageStream<KV<Object, Object>> input1 = appDesc.getInputStream(input4Descriptor); OutputStream<KV<Object, Object>> output1 = appDesc.getOutputStream(output1Descriptor); input1.partitionBy(m -> m.key, m -> m.value, mock(KVSerde.class), "p1").map(kv -> kv).sendTo(output1); }, config); JobGraph jobGraph = (JobGraph) planner.plan(graphSpec); jobGraph.getIntermediateStreams().forEach(edge -> { assertEquals(partitionLimit, edge.getPartitionCount()); }); }
@Test(expected = SamzaException.class) public void testRejectsInvalidStreamStreamJoin() { ExecutionPlanner planner = new ExecutionPlanner(config, streamManager); StreamApplicationDescriptorImpl graphSpec = createStreamGraphWithInvalidStreamStreamJoin(); planner.plan(graphSpec); }
@Test(expected = SamzaException.class) public void testRejectsInvalidStreamTableJoin() { ExecutionPlanner planner = new ExecutionPlanner(config, streamManager); StreamApplicationDescriptorImpl graphSpec = createStreamGraphWithInvalidStreamTableJoin(); planner.plan(graphSpec); }
@Test(expected = SamzaException.class) public void testRejectsInvalidStreamTableJoinWithSideInputs() { ExecutionPlanner planner = new ExecutionPlanner(config, streamManager); StreamApplicationDescriptorImpl graphSpec = createStreamGraphWithInvalidStreamTableJoinWithSideInputs(); planner.plan(graphSpec); }
@Test public void testTriggerIntervalForJoins() { Map<String, String> map = new HashMap<>(config); map.put(JobConfig.JOB_INTERMEDIATE_STREAM_PARTITIONS, String.valueOf(DEFAULT_PARTITIONS)); Config cfg = new MapConfig(map); ExecutionPlanner planner = new ExecutionPlanner(cfg, streamManager); StreamApplicationDescriptorImpl graphSpec = createStreamGraphWithStreamStreamJoin(); ExecutionPlan plan = planner.plan(graphSpec); List<JobConfig> jobConfigs = plan.getJobConfigs(); for (JobConfig config : jobConfigs) { System.out.println(config); } }
@Test public void testTriggerIntervalForWindowsAndJoins() { Map<String, String> map = new HashMap<>(config); map.put(JobConfig.JOB_INTERMEDIATE_STREAM_PARTITIONS, String.valueOf(DEFAULT_PARTITIONS)); Config cfg = new MapConfig(map); ExecutionPlanner planner = new ExecutionPlanner(cfg, streamManager); StreamApplicationDescriptorImpl graphSpec = createStreamGraphWithJoinAndWindow(); ExecutionPlan plan = planner.plan(graphSpec); List<JobConfig> jobConfigs = plan.getJobConfigs(); assertEquals(1, jobConfigs.size()); assertEquals("4", jobConfigs.get(0).get(TaskConfig.WINDOW_MS)); }
@Test public void testTriggerIntervalWithNoWindowMs() { Map<String, String> map = new HashMap<>(config); map.put(JobConfig.JOB_INTERMEDIATE_STREAM_PARTITIONS, String.valueOf(DEFAULT_PARTITIONS)); Config cfg = new MapConfig(map); ExecutionPlanner planner = new ExecutionPlanner(cfg, streamManager); StreamApplicationDescriptorImpl graphSpec = createStreamGraphWithJoinAndWindow(); ExecutionPlan plan = planner.plan(graphSpec); List<JobConfig> jobConfigs = plan.getJobConfigs(); assertEquals(1, jobConfigs.size()); assertEquals("4", jobConfigs.get(0).get(TaskConfig.WINDOW_MS)); }
@Test public void testTriggerIntervalForStatelessOperators() { Map<String, String> map = new HashMap<>(config); map.put(JobConfig.JOB_INTERMEDIATE_STREAM_PARTITIONS, String.valueOf(DEFAULT_PARTITIONS)); Config cfg = new MapConfig(map); ExecutionPlanner planner = new ExecutionPlanner(cfg, streamManager); StreamApplicationDescriptorImpl graphSpec = createSimpleGraph(); ExecutionPlan plan = planner.plan(graphSpec); List<JobConfig> jobConfigs = plan.getJobConfigs(); assertEquals(1, jobConfigs.size()); assertFalse(jobConfigs.get(0).containsKey(TaskConfig.WINDOW_MS)); }
@Test public void testTriggerIntervalWhenWindowMsIsConfigured() { Map<String, String> map = new HashMap<>(config); map.put(TaskConfig.WINDOW_MS, "2000"); map.put(JobConfig.JOB_INTERMEDIATE_STREAM_PARTITIONS, String.valueOf(DEFAULT_PARTITIONS)); Config cfg = new MapConfig(map); ExecutionPlanner planner = new ExecutionPlanner(cfg, streamManager); StreamApplicationDescriptorImpl graphSpec = createSimpleGraph(); ExecutionPlan plan = planner.plan(graphSpec); List<JobConfig> jobConfigs = plan.getJobConfigs(); assertEquals(1, jobConfigs.size()); assertEquals("2000", jobConfigs.get(0).get(TaskConfig.WINDOW_MS)); }