KAFKA-2593 Key value stores can use specified serializers and deserializers

[rhauch]

Add support for the key value stores to use specified serializers and deserializers (aka, "serdes"). Prior to this change, the stores were limited to only the default serdes specified in the topology's configuration and exposed to the processors via the ProcessorContext.

Now, using InMemoryKeyValueStore and RocksDBKeyValueStore are similar: both are parameterized on the key and value types, and both have similar multiple static factory methods. The static factory methods either take explicit key and value serdes, take key and value class types so the serdes can be inferred (only for the built-in serdes for string, integer, long, and byte array types), or use the default serdes on the ProcessorContext.

[asfbot]

kafka-trunk-git-pr #584 FAILURE
Looks like there's a problem with this pull request

[rhauch]

I updated the PR to correct the checkstyle errors.

[asfbot]

kafka-trunk-git-pr #585 SUCCESS
This pull request looks good

[guozhangwang]

LGTM overall, one general comment: could we put various create() functions of different store classes into a single sharable support / factory class, otherwise future users would need to create those functions for each store?

[rhauch]

@guozhangwang, sure, it would be possible to put all of the create methods in a single place. However, I've never been a fan of centralized factory methods because it makes other implementations second-class citizens, since they can't be created using the factory. (Yes, we could use Java's ServiceLoader mechanism, but that's IMO that's overkill and still doesn't allow customization of create methods in other KeyValueStore implementations.)

Now, I do agree that the 2 existing implementations (or 3 including #256) all have similar factory methods with identical signatures, and that might suggest value to a common way to build them. But perhaps a builder (rather than a factory class) might be more usable by processors and reusable by other KeyValueStore implementations. For example:

InMemoryStore<String,Integer> store = InMemoryStore.create(name,context)
                                                   .withKeySerializer(new StringSerializer())
                                                   .withKeyDeserializer(new StringDeserializer())
                                                   .withValueSerializer(new IntegerSerializer())
                                                   .withValueDeserializer(new IntegerDeserializer())
                                                   .withSystemTime()
                                                   .build();

It would be possible to create a base interface and impl for this builder in KeyValueStore that subclasses could reuse and extend with their own methods. The builder could have overloaded methods, too, without resulting in combinatorial explosion of separate factory methods. For example, the builder could have both withSystemTime() and withTime(Time), and the last one used wins. And it could have multiple withKeySerializer methods that take different things, such as a serializer instance, the serializer class (to automatically create the instance using a no-arg constructor), or the key class (for built-in serdes only). IDEs also make builders pretty convenient to use.

This will expand the Kafka Streams code a bit, but IMO it is worth it because it will make it even easier (albeit a bit more verbose) to create KeyValueStore instances in processors.

I'll prototype something up, but should I do that as part of this PR/issue, or under a separate "Simplified approach to create KeyValueStores" issue?

[rhauch]

@guozhangwang, it may be difficult or unattractive to use a builder with the generic key and value types, unless somehow they're specified in the create method or by explicitly typing:

InMemoryKeyValueStore<String, Integer> store = InMemoryKeyValueStore.<String, Integer>create("something", context)
                                               .withKeySerializer(new StringSerializer())
                                               .withKeyDeserializer(new StringDeserializer())
                                               .withValueSerializer(new IntegerSerializer())
                                               .withValueDeserializer(new IntegerDeserializer())
                                               .withSystemTime()
                                               .build();

The good news is that the above code forces the IDE to type-check the with... methods, but it's not pretty. I'll keep trying.

[rhauch]

@guozhangwang, there doesn't appear to be a decent way to use a builder while returning a parameterized type. The example mentioned above is pretty verbose, and IMO the static factory methods on each KeyValueStore implementation class is the most pragmatic way to create instances. It's not ideal, because as you pointed out each class has static methods with identical (or at least nearly identical) signatures. OTOH, I'm not really sure how this might be simplified another way.

BTW, I'd like to pull some of the commits from #256 into this PR, which is really where they belong. Specifically, all but the one commit that creates the InMemoryLRUCacheStore and corresponding test class. I've already moved them over on a local branch and ran a build, so if you're okay with it I'll push the additional commits to this PR and also update #256.

[rhauch]

@guozhangwang, it occurred to me that your goal might actually be to make the implementations of KeyValueStore largely transparent to users, and instead to emphasize that stores can either keep all entries in-memory (and is thus largely limited by available heap) while the other holds entries outside of the heap and this is largely unlimited. Is this accurate?

If this is true, and since builders are relatively difficult to implement for parameterized stores, then I agree that a factory class might be the best way forward. Do you have time to list what you might have been thinking of for an API? Perhaps given some store defined in a Processor:

private KeyValueStore<String, Integer> myStore;

the store could be created in init(...) with something like:

myStore = Stores.inMemory()
                .create(name,context,...);

or with limited memory (see #256):

myStore = Stores.inMemory()
                .maxCount(4096)
                .create(name,context,...);

or to use the database:

myStore = Stores.offHeap()
                .create(name,context,...);

What do you think? This is somewhat like a builder, but all of the parameters that drive the key and value types have to be passed in at once (e.g., only the create(...) methods would use generics).

[rhauch]

@guozhangwang, rebased this PR. The Jenkins build seemed to have failed on core -- is there a way I can resubmit it?

[rhauch]

@guozhangwang, I just added a commit that replaces all of the static factory methods for the different kinds of key-value stores with a simple builder mechanism. Although a fluent API can be used for some parameters, most still have to be passed in to the final 'create(...)' method -- this keeps the API and the implementation very simple. (We could break up more of the parameters into separate fluent-style methods, but this very quickly makes the builder very complicated. See comments above for detail why.)

[rhauch]

Once again, the Jenkins test failures seem irrelevant.

[guozhangwang]

What are the benefits of requiring ProcessorContextImpl implementing the Supplier interface rather than do the casting?

I originally think there will be two categories of users of Kafka Streaming: 1) normal users will only be interfacing the public APIs, 2) some small groups of developers that add windowing def / etc themselves will need to be exposed with the internal implementations in order to code, so it is OK to ask them to know about ProcessorContextImpl. But maybe there are something that I missed here.

[rhauch]

@guozhangwang: The primary benefit is that we can use these components in unit tests using a mock ProcessorContext. Without this small abstraction, all unit tests would have to use a real ProcessorContextImpl and everything that requires.

[guozhangwang]

@rhauch sorry for the late review. I think it's better to keep this PR separate from #256 (but seems it does not contain the commits from that ticket anyways). Overall I think it is good, and I like the Stores.create() mechanism except some minor comment about withTime().

[rhauch]

@guozhangwang, incorporated your comments, rebased upon trunk, successfully ran build locally, and updated this PR.

[rhauch]

@guozhangwang, I've updated the PR to use the Stores.create(...) approach, and the full build passed locally before I pushed. The Jenkins test failure is unrelated to this PR.

[rhauch]

@guozhangwang: Is there anything else you need from me before this can be merged? Once it's merged, I can then rebase #256.

[guozhangwang]

LGTM, thanks @rhauch !