Page History

/**
* CollectionStream provides utilities to build an in memory input stream of collections(list, map). It also supports initialization of 
* multiple partitions for an input stream 
*/
 
public class CollectionStream<T> { 
  // Create an empty stream that a Samza job can produce to
  public static <T> CollectionStream<T> empty(String steamId  
  // Create a stream of messages from input list with multiple partition, key of partitions map is partitionId
  public static <T> CollectionStream<T> of(String streamName, Map<Integer,? extends Iterable<T>> partitions){...}
  
  // 
  public CollectionStream<T> from(CollectionStreamSystem system) {...}
  
  // Create a stream of messages from input list with single partition
  public static <T> CollectionStream<T> of(String steamId, Iterable<T> collection)
}

/**
* Simple Test case using a collection as an input for a low level application. It demonstrates set up and comparison of a test with 
* minimal(eg. here none) configs and it reads an input stream of integers and multiplies each integer with 10
*/
 
// Create a StreamTask
StreamTaskMyStreamTestTask myTask = new StreamTask() {
  @Override
  public void process(IncomingMessageEnvelope envelope, MessageCollector collector, TaskCoordinator coordinator) throws Exception {
    Integer obj = (Integer) envelope.getMessage();
    collector.send(new OutgoingMessageEnvelope(new SystemStream("test-samza","output"), obj*10));
  }
};
 
//CollectionStreamSystem Initializesystem and run the test framework
TestRunner
    .of(myTask)
    .setJobContainerThreadPoolSize(4)
    .addInputStream(CollectionStream.of("test-samza.= CollectionStreamSystem.create("test");


CollectionStream<Integer> input = CollectionStream
	.of("input", {1,2,3,4})
	.from(system);


CollectionStream output =  .addOutputStream(CollectionStream
	.empty("test-samza.output"))
  )
	.from(system);

TestRunner
    .of(MyStreamTestTask.class)
    .addInputStream(input)
    .addOutputStream(output)
    .run();
 
 
// Assertions on the outputs
StreamAssertAssert.that("test-samza.output")assertThat(StreamUtils.getStreamState(output), IsIterableContainingInOrder.contains({10,20,30,20,40})));

/**
* Simple Test case using a collection as an input for a low level application. Itin demonstratesthe set up and comparison of a test with 
* minimal(eg. here none) configs and it reads an input stream of integers and multiplies each integer with 10
*/

HashMap<String,String> configs = new HashMap<String,String>();
 
class MyStreamTaskFactory implements StreamTaskFactory {
    @Override
    public StreamTask createInstance() {
      return (envelope, collector, coordinator) -> {
        return envelope * 10;async mode
*/

// 
public class MyAsyncStreamTask implements AsyncStreamTask, InitableTask, ClosableTask {
  private Client client;
  private WebTarget target; 

  @Override
  public void init(Config config, TaskContext taskContext) throws Exception {
    // Your initialization of web client code goes here
    client = ClientBuilder.newClient();
    target = client.target("http://example.com/resource/").path("hello");
  }

  @Override
  public void processAsync(IncomingMessageEnvelope envelope, MessageCollector collector,
      };
TaskCoordinator coordinator, final TaskCallback callback) }{
   }
}

StreamDescriptor.Input input = StreamDescriptor.<String, String>input("input")
    .withKeySerde(new StringSerde("UTF-8"))
    .withMsgSerde(new StringSerde("UTF-8"));

StreamDescriptor.Output output = StreamDescriptor.<String, String>output("output")
    .withKeySerde(new StringSerde("UTF-8"))
    .withMsgSerde(new StringSerde("UTF-8"));

// InMemorySystem maintains system and stream config
InMemoryTestSystem testingSystem = InMemoryTestSystem.create("samza-test")
    .addInput(input, {1,2,3,4,5})
    .addOutput(output);

 
// JOB Specific Config
configs.put(JobConfig.JOB_NAME(), JOB_NAME);


// Default Single Container configs
configs.putIfAbsent(JobConfig.PROCESSOR_ID(), "1");
configs.putIfAbsent(JobCoordinatorConfig.JOB_COORDINATION_UTILS_FACTORY, PassthroughCoordinationUtilsFactory.class.getName());
configs.putIfAbsent(JobCoordinatorConfig.JOB_COORDINATOR_FACTORY, PassthroughJobCoordinatorFactory.class.getName());
configs.putIfAbsent(TaskConfig.GROUPER_FACTORY(), SingleContainerGrouperFactory.class.getName());


 
StreamTaskApplication app = StreamApplications.createStreamTaskApp(new MapConfig(configs), new MyStreamTaskFactory());
app.addInputs(Collections.singletonList(input)).addOutputs(Collections.singletonList(output)).run();
app.waitForFinish();
 
// Assertions on the outputs
Assert.assertThat(testingSystem.getStreamState(output), IsIterableContainingInOrder.contains({10,20,30,40,50})));

/**
* Simple Test case using a collection as an input for a low level application in the async mode
*/

// 
public class AsyncRestTask implements AsyncStreamTask, InitableTask, ClosableTask {
  private Client client;
  private WebTarget target; 

  @Override
  public void init(Config config, TaskContext taskContext) throws Exception {
    // Your initialization of web client code goes here
    client = ClientBuilder.newClient();
    target = client.target("http://example.com/resource/").path("hello");
  }

  @Override
  public void processAsync(IncomingMessageEnvelope envelope, MessageCollector collector,
      TaskCoordinator coordinator, final TaskCallback callback) {
    target.request().async().get(new InvocationCallback<Response>() {
      @Override
      public void completed(Response response) {
        Integer obj = (Integer) envelope.getMessage();
        collector.send(new OutgoingMessageEnvelope(new SystemStream("test-samza","output"), obj*10));
        callback.complete();
      }

      @Override
      public void failed(Throwable throwable) {
     target.request().async().get(new InvocationCallback<Response>() {
      @Override
      public void completed(Response response) {
        Integer obj = (Integer) envelope.getMessage();
        collector.send(new OutgoingMessageEnvelope(new SystemStream("test-samza","output"), obj*10));
        callback.complete();
      }

      @Override
      public void failed(Throwable throwable) {
        System.out.println("Invocation failed.");
        callback.failure(throwable);
      }
    });
  }

  @Override
  public void close() throws Exception {
    client.close();
  }
}
 

//CollectionStreamSystem Initializesystem and run the test framework
TestRunner
    .of(new AsyncRestTask())
	.setTaskCallBackTimeoutMS(200)
	.setTaskMaxConcurrency(4)
	.addInputStream(CollectionStream.of("test-samza.Input= CollectionStreamSystem.create("test"); 

CollectionStream<Integer> input = CollectionStream
	.of("input", {1,2,3,4}))
	.addOutputStreamfrom(system);

CollectionStream output = CollectionStream
	.empty("test-samza.output"))
	output")
	.from(system);

TestRunner
    .of(MyAsyncStreamTask.class)
    .addInputStream(input)
    .addOutputStream(output)
    .run();
 
// Assertions on the outputs
StreamAssert.that("test-samza.output").containsInAnyOrder({10,20,30,40});