MINOR: Add StreamsConfig to TopologyTestDriver constructor to enable ease of testing

[wlsc]

This PR offers a simple addition to enable a possibility to pass a StreamsConfig instance for the TopologyTestDriver.

Why this was offered? In our project we have found very useful to create a StreamsConfig instance with all injected dependencies (by Spring) and then still be able to create TopologyTestDriver, which is now possible with this PR.

P.S. Sorry for so many commits -- I should have merged it first and then create a branch.

[wlsc]

@mjsax could you please take a look into it? :)

[mjsax]

Thanks for the PR @wlsc. Note, this would be a public API change and requires a KIP. It's not a MINOR change.

Also note that KafkaStreams constructor accepting StreamsConfig is deprecated -- in general, it seems easier for users to pass in a Properties object -- creating a StreamsConfig object seems to be boiler plat code.

Can you elaborate why it's easier via Spring to inject configs in StreamsConfig instead of using a Properties object? I am not against the change, but what to understand better if it's required or not.

[wlsc]

@mjsax I use Spring and overwrite getConfiguredInstance inside StreamsConfig to provide Kafka Streams with a correct object.

The way, in which Kafka Streams creates implementors of the interface DeserializationExceptionHandler, leaves no control options over creation and pass over of an instance to Kafka Streams.

Therefore, in case there are dependencies to the implementor of the DeserializationExceptionHandler, Kafka Streams fails to start. A use case example could be the following: I would like to report all these errors to some external service.

In this regard, I would suggest applying similar logic as for setUncaughtExceptionHandler for the KafkaStreams class.

It would be great if there were an opportunity to have control over creation of DeserializationExceptionHandler as well as of other handlers.

[mjsax]

@wlsc Thanks for your reply. I am not sure if I understand completely. Why do you need to overwrite getConfiguredInstance()? As DeserializationExceptionHandler implements Configurable, it will call configure() that should allow you to execute custom code on object creation. It's unclear to me, why this is not sufficient?

[vvcephei]

Hi @wlsc ,

It seems surprising that it's not possible to inject the desired Properties object. Can you provide a small example program demonstrating the problem(s)?

Thanks,
-John

[wlsc]

@vvcephei The problem is not in the Properties injection itself.
I've created a demo project, which demonstrates how we are forced now with 2.0.0 version of Kafka Streams API (StreamsConfig is deprecated for KafkaStreams creation) to supply only Properties for Kafka Streams. This also includes a demo code of a version which is more Spring-fiendly. In that version as I have described above, I am overriding getConfiguredInstance while creating StreamsConfig instance.

demo.zip -- please take a look at both usages in two separated packages.

[vvcephei]

Hey @wlsc , sorry; I had a distracting week. I'll look at it tomorrow. Thanks for the example!

[vvcephei]

Ok, I've looked through your example, and I think I understand now.

I think that we interpreted the comments as saying that you are not able to construct the Streams app using Spring. But it seems like what you were actually saying was that you prefer to let Spring construct the exception handler (for example) rather than having Streams do it.

Just to be super clear, the situation is currently that you register the exception handler class, and Streams uses reflection to invoke the 0-arg constructor and passes the config to the new instance's "configure" method. Thus, if your exception handler needs some other dependency, you have to construct it ahead of time and insert it into the config Properties, and then get it back out inside the configure method of your exception handler.

Your preference would be to override part of the StreamsConfig class to replace the reflection-based instantiation with Spring's mechanism. This yields two advantages:

1. Spring can search the classpath to satisfy the dependency, so you don't need to construct it and register it in the config ahead of time.
2. Spring can invoke other constructors besides the 0-arg one, so you can "configure" the handler at construction time, rather than at configuration time. The primary benefit here seems to be that you can store the dependencies in final fields.

This seems reasonable to me. Essentially, making the construction of configured classes pluggable would let anyone plug in Spring or Guice or whatever and get more featureful dependency injection than the default, reflection/configure, mechanism.

Sorry it took so long to digest what you were getting at.

That said, this change does need a KIP, since it alters the public interface. This is actually a good thing, since it gives you the opportunity to propose not only adding this constructor to TopologyTestDriver but also de-deprecating the KafkaStreams constructor you're using.

I can also imaging the community wanting to discuss if there's a more ergonomic approach than overriding getConfiguredInstance, and also if there are any similar changes we'd consider making to Producer and Consumer.

[vvcephei]

The KIP process is documented here: https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Improvement+Proposals. Let us know if you need any help on it!

[mjsax]

@wlsc What is the status of this? Are you planning to do a KIP?

[wlsc]

@mjsax I have a great interest in writing KIP. I am not sure what would be a timeline, since It would be my first time writing a proposal like this. If there any deadlines I should be aware of, please let me know.

[vvcephei]

Hey @wlsc ,
The upcoming release is 2.1. Here's the release plan: https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=91554044

The KIP deadline is 24 Sept., so your kip would need to be voted and accepted by then. The vote itself takes at least 72 hours, per the KIP rules (https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Improvement+Proposals).

This might be a little tight to write your first kip, guide the discussion, and get the vote passed. But I wouldn't stress about it. If you don't get it in for 2.1, there's always 2.2+.

I'd recommend to peruse a few recent Streams KIPs (https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Streams) to get a feel for the pattern and just copy one as a template.

Let us know if you have any questions!

(Just something to think about as you design the KIP):
I'm wondering if it would be more ergonomic just to register a ConfiguredInstanceFactory instead of having to subclass StreamsConfig, like this:

public interface ConfiguredInstanceFactory {
    public <T> T getConfiguredInstance(String key, Class<T> type);
}

and then add a new StreamsConfig constructor for it:

new StreamsConfig(Properties config, ConfiguredInstanceFactory configuredInstanceFactory)

The main advantages here:

• It actually advertises the use case (it otherwise wouldn't be obvious that you can plug in your own DI framework by subclassing StreamsConfig)
• you have a much lower risk of accidentally altering the behavior of StreamsConfig in unexpected ways

[wlsc]

@vvcephei @mjsax hi. I've created a KIP for this and started discussion in the mailing list.

[guozhangwang]

@wlsc let's continue on the discussion thread of the KIP, and hopefully we can move forward to the voting thread soon. Personally I'm in favor of the current approach in this PR but we can keep our discussion on the mailing list.

[mjsax]

Closing this stale PR due to inactivity. IIRC, we never received a KIP for this change, what would be required for a public API change.