Status

Current state: [ UNDER DISCUSSION ]Accepted

Discussion thread: Add in-memory system

...

• Approach A - Use existing BlockingEnvelopeMap and have one common class that shares the responsibility of consumer as well as producer. The class will be responsible for handling both producing and consuming messages off the same queue.
• Approach B - Have separate producer and consumer. Tie up the consumer with the producer so that producer has hooks to produce to the same underlying `BlockingEnvelopeMap` that consumer uses.
• Approach C - Have separate consumer and producer. Introduce a custom queue that are shared between consumer and producer. The queue lifecycle is managed by the SystemAdmin.

...

• .

I am leaning towards Approach C as its simpler, not tied to BlockingEnvelopeMap and has separation of concerns.

Example Usages

High level stateless application

/**
 * Sample test case w/ multiple input partition using collection based system.
 */
 
...
...
 
ImmutableList<IV> inputA = ...
ImmutableList<IV> inputB = ...
 
List<OV> outputData = ...   // mutable
 
StreamDescriptor.Input<IK,IV> input = StreamDescriptor.<IK,IV>input("test-stream")
       .from(ImmutableList.of(inputA, inputB));
 
StreamDescriptor.Output<OK,OV> output = StreamDescriptor.<OK,OV>output("output-test-stream")
       .from(outputData);
 
// application logic
StreamApplication app = StreamApplication.create(...);
app.from(input)
   .map(...)
   .sendTo(output);
 
app.run();
app.waitForFinish();
 
// assertions on outputData

High level application with durable state

/**
 * Sample test case using collection based system for high level application with durable state.
 * Note: We don't provide the ability to recover data/bootstrap data for store using in-memory system. It should work seamlessly once we have StoreDecriptor.
 */
 
...
...
 
ImmutableList<IV> inputA = ...
 
List<OV> outputData = ...   // mutable
 
StreamDescriptor.Input<IK,IV> input = StreamDescriptor.<IK,IV>input("test-stream")
       .from(ImmutableList.of(inputA));
 
StreamDescriptor.Output<OK,OV> output = StreamDescriptor.<OK,OV>output("output-test-stream")
       .from(outputData);
 
// application logic
StreamApplication app = StreamApplication.create(...);
app.from(input)
   .map(...)
   .sendTo(output);
 
app.run();
app.waitForFinish();
 
// assertions on outputData
 

Low level stateless application

/**
 * Sample test case using collection based system for low level application.
 */

...
...

ImmutableList<IV> inputA = ...

List<OV> outputData = ... // mutable

StreamDescriptor.Input<IK,IV> input = StreamDescriptor.<IK,IV>input("input-stream-low-level-app")
	.from(ImmutableList.of(inputA));
StreamDescriptor.Output<OK,OV> output = StreamDescriptor.<OK,OV>output("output-stream-low-level-app")
	.from(outputData);

// application logic
StreamTaskApplication app = StreamTaskApplication.create(config, new MyTaskFactory())
	.addInputs(Collections.singletonList(input))
	.addOutputs(Collections.singletonList(output));

app.run();
app.waitForFinish();

// assertions on outputData

Low level application with durable state

/**
 * Sample test case using collection based system for low level application with durable state.
 * Note: We don't provide the ability to recover data/bootstrap data for store using in-memory system. It should work seamlessly once we have StoreDecriptor.
 */

...
...

ImmutableList<IV> inputA = ...

List<OV> outputData = ... // mutable

StreamDescriptor.Input<IK,IV> input = StreamDescriptor.<IK,IV>input("input-stream-low-level-app")
	.from(ImmutableList.of(inputA));
StreamDescriptor.Output<OK,OV> output = StreamDescriptor.<OK,OV>output("output-stream-low-level-app")
	.from(outputData);

// application logic
StreamTaskApplication app = StreamTaskApplication.create(config, new MyTaskFactory())
	.addInputs(Collections.singletonList(input, changelog))
	.addOutputs(Collections.singletonList(output));

app.run();
app.waitForFinish();

// assertions on outputData

Low level application with custom IME 

/**
 * Sample test case using collection based system for low level application using custom IME.
 * It demonstrates the use of IME as a data source as opposed to raw message. The users are responsible for creating a 
 * complete IME object with partition information.
 */
 
...
...
 
ImmutableList<MyIME> inputData = Utils.createMyIME(...);
 
List<OV> outputData = ... // mutable
 
StreamDescriptor.Input<Object, Object> input = StreamDescriptor.<>input("input-stream-low-level-app")
	.from(inputData);
StreamDescriptor.Output<OK,OV> output = StreamDescriptor.<OK,OV>output("output-stream-low-level-app")
	.from(outputData);
 
// application logic
StreamTaskApplication app = StreamTaskApplication.create(config, new MyTaskFactory())
	.addInputs(Collections.singletonList(input, changelog))
	.addOutputs(Collections.singletonList(output));

app.run();
app.waitForFinish();

// assertions on outputData

Low level application with manual checkpoint

Unsupported Use cases

For In V1, we don't support checkpointing

...

...

will not support the following use cases since it has a depdencies.

• High level application with durable state
• Low level application with durable state
• Application with manual checkpoint. Note. Manual checkpointing will result in a no-op and might not result in desired behaviour.

...

Samza SQL application

Users should be able to leverage in-memory collection based system to test Samza SQL application provided Samza SQL integrates with SEP-2.

...