...

1. Allow the current thread access to other threads on standby. And when a need for increased processing speed is necessary, we could call upon these reserve threads and then release them once their task is complete so that they could be used elsewhere.
2. Another option instead is to maintain the isolation of the threads, and use a method that does not block (i.e. send a request with a callback which is only called if the records still fail – it would act like a feedback loop and send a request againuse a similar policy to KafkaConsumer#commitAsync).

Public Interfaces

We have a couple of choices available to us as mentioned in the ticket discussion (KAFKA-6989). Recall that in asynchronous methods, we do not need inputs from a previous function to start it. All we need is the signal. However, when offsets have been committed or processed in an asynchronized manner, we will need to determine the behavior that occurs after the conclusion of the process. It would also be wise to remember that asynchronous processing does not necessarily require an extra thread.  When offsets are committed or received, we should consider the following:

...

Now we have two ways to approach this issue. Note that in KafkaConsumer, its implementation basically prevents more than one thread from calling it at once. So if we wish to follow a similar methodology with Kafka Streams, we will have a couple of tradeoffsthe following:

1. Positive sides: Ordering would still be guaranteed, and exactly once will still be the same. We would add a method(s) with behavior similar to KafkaConsumer#commitAsync in that it accepts a callback defined by the user and calls it once it is done. The user's thread will move on while a hidden process by Kafka takes overwe construct a future that waits on the task's completion. In this manner, any new methods does not block.
2. Negative sides: As mentioned, failure handling would still be a major problem (the user has to deal with this). We could still end up stuck on one record which for some reason continues to stubbornly fail. It could end up slowing down the whole process. 

So basically, what would this does is that it allows the user to choose between a method that does not block and one that does. The isolation between threads are still maintainedNote that this line of thought does not change much because this is close to saying: you could choose between KafkaConsumer#commitSync and KafkaConsumer#commitAsync(). Next, we have the multithreaded approach and we can take it one step furtheran alternative route. Instead of one thread, we could have the followingsome more tradeoffs:

1. Positive sides: Failure handling is better now in that multiple threads are on the job. While the a secondary thread takes care of the failed metadata, the primary thread could move on processing new ones. Since the failed metadata topic's workload is not constantly increasing, we will have time to process them. Once the secondary thread has finished with the failed records, it could be terminated, thus freeing up CPU resources and space. Latency would be reduced.
2. Negative sides: Ordering is now impossible to guarantee, as is exactly-once because we have no way of knowing which records has been returned since asynchronous processes have no way of communicating between one another. .

In the first approach I outlined, we are essentially giving the user some more flexibility in deciding how to resolve the latency and failure handling problem (that is they can choose on whether not to sacrifice ordering for latency). They would have to implement the multithreaded portion themselves to make up for it (and we should be able to infer that most clients are not up for this kind of effort)  The second approach takes some load off the client's back in that we figure out how to process the records using multiple threads, and clients doesn't have to worry about anything complex. Note that with the second plan, we still add the capability to commit synchronously or asynchronously regardless.no futures would be involved as secondary threads would be processing it directly using blocking methods (like KafkaConsumer#commitSync). 

We could set the first approach as the default policy, while the second approach (which is the one that Samza uses) would be used only if the client specifies us to use more than one thread per stream task.   

Proposed Changes

For both the first and second plan, we will would have to add some capability to process asynchronously (even if you are using only one thread). So we would include a method such as the following:some new methods that might look something like this: 

/** Retrieves record metadata for a topic asynchronously. This method does not block.
 *  @param callback will be called when the records has been retrieved or has failed.
 * /
T <T> getMetadtaAsync(); @param storeName the name of the store the user is querying.
 * /
void getMetadataAsync(RecordCallback callback, String storeName);

The purpose of RecordCallback is analogous to KafkaConsumer's OffsetCommitCallback. The user would use this interface (or class?) to define the end behavior of a method.  It could be used to notify the user that they could move on to processing the next record, or it could be used as a feedback loop to send another request for the same record (if it failed).

In the second plan, we will have to include a new config such as num.threads.per.task to help us ascertain the number of threads, maximum, that could be used per task. These threads would be simultaneously processing from the same task (however, ordering would no longer be guaranteed).

If num.threads.per.task is equal to one, then we go to our default policy. Otherwise, we would go with the second one.

Compatibility, Deprecation, and Migration Plan

...

There might be some problems if a new KafkaStreams instance is brought online in which it gives the offsets out of order when the user is expecting it to be in-order. So in order to maintain the old behavior, we will keep the current structure of Kafka Streams methods intact (although some of its statements might have to be tweaked to acommadate for the new change). 

Rejected Alternatives

If there are alternative ways of accomplishing the same thing, what were they? The purpose of this section is to motivate why the design is the way it is and not some other way.