Page History

/**
* TestRunner provides static factory for quick setup of Samza environment for testing Low level api and High level api , users can configure input streams 
* they consume from and output streams they produce to, users pass in their Samza job api logic via StreamTask and /AsyncStreamTask/StreamApplication
*/ 

public class TestRunner {
  
  // Static Factory to config & create runner for low level api 
  public static TestRunner of(StreamTask task) {...}
  public static TestRunner of(AsyncStreamTask task) {...}
  public static TestRunner of(StreamApplication app) {...}
  
  // Add/Ovveride any custom configs
  public TestRunner addOverrideConfigs(Map<String,String> configs) {...}
  public TestRunner addOverrideConfigs(String configUri) {...}
  public TestRunner addOverrideConfig(String key, String val){...}
 
  // Set container mode either single container or multi container
  public TestRunner setContainerMode(Mode mode) {...}
  
  // Multithreading configs for users
  public TestRunner setTaskMaxConcurrency(Integer value) {...}
  public TestRunner setTaskCallBackTimeoutMS(Integer value) {...}
  public TestRunner setJobContainerThreadPoolSize(Integer value) {...}
 
  // Configure state for application
  public TestRunner addState(String storeName) {...}  
  
  // Configure an input stream for samza system, that taskjob can consume from
  public TestRunner addInputStream(CollectionStream stream) {...}
  // Configures an output stream for samza system, that taskjob can producer to
  public TestRunner addOutputStream(CollectionStream stream) {...}
  
  // Run the app
  public void run() {...}
}

 

/**
* 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 taskjob can produce to
  public static <T> CollectionStream<T> empty(String steamId) {...}
  
  // Create a stream of messages from input list with single partition
  public static <T> CollectionStream<T> of(String steamId, Iterable<T> collection){...}
  
  // Create a stream of messages from input list with multiple partition, key of partitions map is partitionId
  public static <T> CollectionStream<T> of(String steamId, Map<Integer,Iterable<T>> partitions){...}
  
}