StreamApplicationDescriptorImpl extends ApplicationDescriptorImpl<StreamApplicationDescriptor> implements StreamApplicationDescriptor { @Override public <M> OutputStream<M> getOutputStream(OutputDescriptor<M, ?> outputDescriptor) { addOutputDescriptor(outputDescriptor); String streamId = outputDescriptor.getStreamId(); Serde serde = outputDescriptor.getSerde(); KV<Serde, Serde> kvSerdes = getOrCreateStreamSerdes(streamId, serde); boolean isKeyed = serde instanceof KVSerde; outputStreams.put(streamId, new OutputStreamImpl<>(streamId, kvSerdes.getKey(), kvSerdes.getValue(), isKeyed)); return outputStreams.get(streamId); } 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(expected = IllegalArgumentException.class) public void testGetOutputStreamWithNullSerde() { String streamId = "test-stream-1"; GenericSystemDescriptor sd = new GenericSystemDescriptor("mockSystem", "mockSystemFactoryClass"); GenericOutputDescriptor osd = sd.getOutputDescriptor(streamId, null); new StreamApplicationDescriptorImpl(appDesc -> { appDesc.getOutputStream(osd); }, getConfig()); }
@Test(expected = IllegalStateException.class) public void testSetDefaultSystemDescriptorAfterGettingOutputStream() { String streamId = "test-stream-1"; GenericSystemDescriptor sd = new GenericSystemDescriptor("mockSystem", "mockSystemFactoryClass"); GenericOutputDescriptor osd = sd.getOutputDescriptor(streamId, mock(Serde.class)); new StreamApplicationDescriptorImpl(appDesc -> { appDesc.getOutputStream(osd); appDesc.withDefaultSystem(sd); }, getConfig()); }
@Test(expected = IllegalStateException.class) public void testGetSameOutputStreamTwice() { String streamId = "test-stream-1"; GenericSystemDescriptor sd = new GenericSystemDescriptor("mockSystem", "mockSystemFactoryClass"); GenericOutputDescriptor osd1 = sd.getOutputDescriptor(streamId, mock(Serde.class)); GenericOutputDescriptor osd2 = sd.getOutputDescriptor(streamId, mock(Serde.class)); new StreamApplicationDescriptorImpl(appDesc -> { appDesc.getOutputStream(osd1); appDesc.getOutputStream(osd2); }, getConfig()); }