Introduce functionality for kafka functionality for tempo

Currently tempo has clients that talk to remote models. In order to introduce support for stream processing such as through Kafka, we would need to extend the current interfaces to support asynchronous processing.

Models would be defined the same way

sklearn_model = Model(
    name="test-iris-sklearn",
    platform=ModelFramework.SKLearn,
    local_folder=f"{artifacts_folder}/{SKLearnFolder}",
    uri="s3://tempo/basic/sklearn",
)

xgboost_model = Model(
    name="test-iris-xgboost",
    platform=ModelFramework.XGBoost,
    local_folder=f"{artifacts_folder}/{XGBoostFolder}",
    uri="s3://tempo/basic/xgboost",
)

Option 1: AsyncIO

Each instance to a pipeline is processed synchronously, with each subsequent model "synchronously", of course using AsyncIO to release control, but waiting until the respective model "returns" the output to the output_topic.

Advantages

syntax wouldn't change for requests based vs stream based processing.
State is maintained, which means that "combiner" logic is orders of magnitude simpler (or even feasible)

Disadvantages

The logic to await for the model to respond is not trivial with Python + Kafka.
The models would be dependent on the availability of processing as opposed to being fully async (eg could run forever or time out logic needs to be introduced)

@pipeline(
    name="classifier",
    uri="s3://tempo/basic/pipeline",
    local_folder=f"{artifacts_folder}/{PipelineFolder}",
    models=PipelineModels(sklearn=sklearn_model, xgboost=xgboost_model),
)
def classifier(payload: np.ndarray) -> Tuple[np.ndarray, str]:
    res1 = await classifier.models.sklearn(input=payload)

    if res1[0] == 1:
        return res1, SKLearnTag
    else:
        return await classifier.models.xgboost(input=payload), XGBoostTag

Option 2: Fully Asynchronous Streams

The second option is to approach it in the fully traditional independent stream processing approach, where each independent data instance would be processes as single step across the streaming pipeline process.

Advantages

Each data point is processed asynchronously which means more efficient use of compute

Disadvantage

Different syntax
No state locally

p = Pipeline(
    name="classifier",
    uri="s3://tempo/basic/pipeline",
    local_folder=f"{artifacts_folder}/{PipelineFolder}",
    models=PipelineModels(sklearn=sklearn_model, xgboost=xgboost_model),
)

@p.pipeline
def classifier(payload: np.ndarray) -> Tuple[np.ndarray, str]:
    classifier.models.sklearn.send(input=payload)

@p.models.sklearn
def sklearn(res1):
    if res1[0] == 1:
        return res1, SKLearnTag
    else:
        p.models.xgboost.send(input=payload), XGBoostTag

@p.models.xgboost
def xgboost(res):
    return res

SeldonIO / tempo