Status
Current state: [ UNDER DISCUSSION ]
Discussion thread: <link to mailing list DISCUSS thread>
JIRA:
Released:
Problem:
Samza today provides various api's like high level (fluent/application) for chaining complex jobs, low level (task) with configuration like Samza-Yarn, Samza as a library (Samza standalone) and in coming future Samza will support In-memory system for in memory streams that a user can produce/conusme from. Still Samza users have to get their hands dirty with too much configuration details and dependency on external system(like kafka) to write integration tests. In order to alleviate this we intend to provide a fluent test api to users. This will help in setting-up quick tests against their system without depending on underlying incoming stream e.g Kafka Topic. The goal of the proposed SEP we will provide a standardized framework to set-up continuous integration test for High Level Api, Low Level Api with Single Container configuration & Multi Container Configuration (Samza standalone using Zookeeper) with ease. Users won’t have to worry about the underlying Samza config details or even understand implementation details. This abstraction will increase developer productivity and help client's make their Samza system robust.
Motivation
Addition of this Continuous Integration Test Framework will alleviate:
Lack of a standardized way to set up integration tests for users
Lack of brevity in code and much detailed understanding low level config details of the Samza Api to set up samza systems
Lack of a pluggable Test System with any Samza single & multi-container configuration
Set of robust tests for current Samza api
Assumptions
System depends on In Memory system to spin up in-memory system consumers and producers for in memory data streams
Single Container mode runs on SingleContainerGrouperFactory, that means these test are supposed to be ran on a single container environment
Multi Container system leverages Samza as a library (Samza standalone) using Zookeeper as coordination service for setting up test
In order to provide a testing api with least config management, basic configs are set up for users, but flexibility to add/override some custom config is also provided
Testing is always supposed to be done using bounded streams using EndOfStreamMessage
Proposed Changes
We propose a three step process for configuration of integration test for your samza job. Three facets are data injection, transformation and validation.
- Data Injection phase means configuring the input source for Samza job.
- Data Transformation phase means api logic of samza job(low-level/high-level).
- Data Validation phase asserts the expected results to the actual results computed after running a Samza job.
Data Injection:
In the context of this Test framework an input stream means a stream that Samza job may consume from and an output stream means a stream that Samza job can produce to. The initial source of data stream is always supposed to be bounded since this is a test, and that may originate as one of the following sources listed below.
- In Memory Data Stream
Users have ability to produce to & consume from in memory system partitions after SEP-8. In Memory Data system alleviates the need of serialization/deserialization of data since it does not persist data. We take advantage of this and provide very succinct Stream classes to serve as input data sources, which are the following:- Collection Stream
Users can plug a collection (either List or Map) to create an in-memory input stream
e.g: CollectionStream.of(...,{1,2,3,4}) - Event Builder Stream
Event builder helps a user to mimic runtime samza processing environment in its tests, for example adding an exception in the steam, advancing time for window functions.
- Collection Stream
- File Stream
Users can create an input stream reading and writing from local file
e.g: FileStream.of("/path/to/file") - Local Kafka Stream
Users can also consume bounded streams from a kafka topic which serves as initial input. Samza already provide api's to consume from and produce to kafka. For kafka streams we will need serde configurations
For in memory streams the api actually initializes the stream by spinning up a Samza producer using an InMemorySystemProducer to write the stream, this is how a collection of data or events gets initialized as a steam. It also configures any output stream if the user has added any.
Data Transformation:
This is the Samza job you write, this can be either done using Low Level Api or the fluent High Level Api. Test frameworks provides api to set up test for both the cases. Test framework supports both the apis with Single container and Multi-container mode. Users implement StreamTask and Async Stream task in the same way, as they do for their Samza job. For high level api users don't need a class implementing StreamApplication, they just configure message streams of any type and apply operators on it directly (see the sample example below).
Data Validation:
Data Types & Partitions:
The framework will provide complete flexibility for usage of primitive and derived data types. Serdes are required for local kafka stream and file stream but in memory streams dont require any Serde configuration. Test framework will provide api's for initialization of input streams (both single and multi-partition), and also data validation on single partition and multi-partition of the bounded streams
Running Config
Traditionally we ask users to set up config for any samza job, for test purposes we set up basic config boiler plate for users and provide them a flexible option to still add any custom config (rarely needed), api exposes functions to configure single container or multi container mode (using Zookeeper). It also provides functions to configure concurrency semantics for their Samza job.
Future Changes with Stream Descriptors:
The test framework is designed in a way which asks users to do none or minimal Samza configs, in future we intend to use StreamDescriptors in the test framework to do Samza configs. This change would cause a small change in the way user passes their custom configs (if any) to the test framework
Public Interfaces
Two Apis for writing tests are: Low Level Test Api (TestTask) & High Level Api (TestApplication)
/** * TestTask provides static factory for quick setup of Samza environment for testing low 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 */ public class TestTask { private TestTask(String systemName, StreamTask task, HashMap<String, String> config) {...} private TestTask(String systemName, AsyncStreamTask task, HashMap<String, String> config) {...} // Static Factory to config & create runner for low level api public static TestTask create(String systemName, StreamTask task, HashMap<String, String> config) {...} public static TestTask create(String systemName, AsyncStreamTask task, HashMap<String, String> config) {...} // Multithreading configs for users public TestTask setTaskMaxConcurrency(Integer value) {...} public TestTask setTaskCallBackTimeoutMS(Integer value) {...} public TestTask setTaskMaxConcurrency(Integer value) {...} // Configure an input stream for samza system, that task can consume from public TestTask addInputStream(CollectionStream stream) {...} // Configures an output stream for samza system, that task can producer to public TestTask addOutputStream(CollectionStream stream) {...} // Run the app public void run() {...} }
/** * TestApplication provides static factory for quick setup of Samza environment for testing High Level Api, users can configure input * streams and then apply various operators on streams and run the application */ public class TestApplication<T> { private TestApplication( HashMap<String, String> configs, Mode mode) // Static factory to config & create runner for High level api public static TestApplication create(String systemName, HashMap<String, String> config) {...} // Configure any kind of input stream for samza system, and get the handle of message stream to apply operators on public <T> MessageStream<T> getInputStream(EventStream<T> stream) {...} public <T> MessageStream<T> getInputStream(CollectionStream<T> stream) {...} public <T> MessageStream<T> getInputStream(FileStream<T> stream) {...} // Run the app public void run() { }; }
Types of Input Streams
/** * 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> { private CollectionStream(String systemStream, List<T> collection, Integer partitionCount) {...} private CollectionStream(String systemStream) {...} // Create an empty stream that a Samza task can produce to public static <T> CollectionStream<T> empty(String systemStream) {...} // Create a stream of messages from input list with single partition public static <T> CollectionStream<T> of(String systemStream, List<T> collection){...} // Create a stream of messages from input list with multiple partition public static <T> CollectionStream<T> ofPartitions(String systemStream, List<List<T>> collection){...} // Create a stream of messages from input list with multiple partition public static <K, V> CollectionStream<KV<K, V>> of(String systemStream, Map<K, V> elems) {...} }
/** * EventStream provides utilities to build an in memory input stream of events. It helps mimic run time environment of your job, * advancing * time for windowing functions */ public class EventStream<T> { public static abstract class Builder<T> { public abstract Builder addElement(); public abstract Builder addException(); public abstract Builder advanceTimeTo(long time); public abstract EventStream<T> build(); } public static <T> Builder<T> builder() {...} }
/** * FileStream provides utilities to build a stream of messages from a file on disk */ public class FileStream<T> { public static <T> FileStream<T> of(String fileUri) {...} }
A Sample Test:
/** * Sample 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 StreamTask 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)); } }; // Initialize and run the test framework TestTask .create(systemName: "test-samza", myTask, config, mode.SINGLE_CONTAINER) .setJobContainerThreadPoolSize(4) .addInputStream(CollectionStream.of(streamName: "input", {1,2,3,4})) .addOutputStream(CollectionStream.empty(streamName: "output")) .run(); // Assertions on the outputs TaskAssert.that("test-samza", "Output").containsInAnyOrder({10,30,20,40});
// Initialize and run the test framework TestApplication .create(config, mode.SINGLE_CONTAINER); .getInputStream(CollectionStream.of(streamName: "input", {1,2,3})) .map(s -> "processed " + s) .run(); // Assertions on the outputs StreamAssert.that(stream).containsInAnyOrder(Arrays.asList("processed 1", "processed 2", "processed 4"));
Implementation and Test Plan
- Introduce the new interface and api for test framework
- Use the same api to write comprehensive integration test for:
- Samza Low level Async Api (Single and Multi-Container Mode)
- Samza Low level Synchronous Api (Single and Multi-Container Mode)
- Samza High Level Api (Single and Multi-Container Mode)
Compatibility, Deprecation, and Migration Plan
As this is a new feature, no plans are required for compatibility, deprecation and migration.
Rejected Alternatives