KAFKA-3452 Follow-up: Optimize ByteStore Scenarios by guozhangwang · Pull Request #2333 · apache/kafka

guozhangwang · 2017-01-09T00:50:04Z

This is a refactoring follow-up of #2166. Main refactoring changes:

Extract InMemoryKeyValueStore out of InMemoryKeyValueStoreSupplier and remove its duplicates in test package.
Add two abstract classes AbstractKeyValueIterator and AbstractKeyValueStore to collapse common functional logics.
Added specialized BytesXXStore to accommodate cases where key value types are Bytes / byte[] so that we can save calling the dummy serdes.
Make the key type in ThreadCache from byte[] to Bytes, as SessionStore / WindowStore's result serialized bytes are in the form of Bytes anyways, so that we can save unnecessary Bytes.get() and Bytes.wrap(bytes).

Each of these should arguably be a separate PR and I apologize for the mess, this is because this branch was extracted from a rather large diff that has multiple refactoring mingled together and @dguy and myself have already put lots of efforts to break it down to a few separate PRs, and this is the only left-over work. Such PR won't happen in the future.

Ping @dguy @enothereska @mjsax for reviews

asfbot · 2017-01-09T00:51:08Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/606/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-01-09T00:51:19Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/607/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-01-09T00:53:42Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/608/
Test FAILed (JDK 8 and Scala 2.11).

…f logged stores beforehand

asfbot · 2017-01-09T04:41:54Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/611/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-01-09T04:41:55Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/612/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-01-09T04:42:02Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/610/
Test FAILed (JDK 7 and Scala 2.10).

guozhangwang · 2017-01-09T04:57:24Z

@dguy @mjsax for reviews (this is still WIP):

Follow the refactoring ideas in KAFKA-3452: Support session windows #2166 (review). Now the hierarchies can be viewed in RocksDBKeyValueStoreSupplier, RocksDBWindowStoreSupplier, RocksDBSessionStoreSupplier. Note that when caching or logging is enabled, all the underlying stores will be typed as <Bytes, byte[]> for performance optimizations.
Added a few AbstractXXX classes to collapse common functions across some peer classes.
Refactored Stores a bit to better aligned with the supplier class.
Removed some redundant inner maps in ProcessorTopology which is only used ProcessorStateManager for flushing stores and remembering the changelog topic of the stores, instead let the ProcessorTopology to construct the full map from stores to their corresponding changelog topics if they are logging enabled within build(). It also allows to extract logging out of RocksDB.

Some more thoughts

With WrapperStateStore, we can now call context.register(..) in the outer-most store and call its inner().put(...) in the restoreCallback instead of calling it in the inner-most store, and hence can further get rid of the root parameter.
We can let users-customized aggregation / join operators to be per-store configurable so that we can also remove the TupleForwarder interface. I.e.:

StateStoreSupplier<...> mySupplier = new MyStateStoreSupplier(...)

// this means library will not add any wrappers for logging / caching / etc
// if users have caching mechanism in its only supplier, she needs to consider 
// flushing / eviction / forwarding herself.
stream.groupByKey(...).aggregate(..., supplier);   

// this means library will wrap user-provided underlying store engine as required
// users do not need to consider the above and our wrapper will do that without hacking forwarder.
StateStoreSupplier<...> myWrappedSupplier = Stores.create(storeName)
                .withKeys(keySerde)
                .withValues(aggValueSerde)
                .persistent()
                .enableCaching();
                .wrap(mySupplier);
                .build();

stream.groupByKey(...).aggregate(..., supplier);

dguy · 2017-01-09T11:36:49Z

+        InMemoryWrapperFactory<K, V> disableLogging();
+    }
+
+    private static abstract class AbstractInMemoryFactory<K, V> implements InMemoryWrapperFactory<K, V> {


I'm not sure this is necessary? The only sub-class is InMemoryKeyValueFactory.

dguy · 2017-01-09T11:41:34Z

-        StateStoreSupplier build();
+        PersistentWrapperFactory<K, V> disableCaching();
+    }
+


I don't think this should be an inheritance hierarchy. This class seems like it should be a standalone class that is used to collect the decisions/conifg etc for logging and caching. It could be constructed and passed in to what are currently its subclasses.
With inheritance we end up with competing interfaces in the hierarchy, i.e., the same methods on PersistentWrapperFactory and PersistentKeyValueFactory

dguy · 2017-01-09T11:43:09Z

+            this.valueSerde = valueSerde;
+        }
+
+        @Override


This is an example of how the inheritance doesn't really work. It actually needs to return this. I think it would be much better to get rid of the inheritance and use delegation instead.

dguy · 2017-01-09T12:47:38Z

@guozhangwang - i've made a quick pass to try and understand where this is heading. I left a couple of comments, but i largely decided not to comment too much as this is a WIP and i wasn't sure what you are after.

Anyway, i think the refactoring of the StateStore hierarchy generally looks good. The introduction of the WrapperStateStore interface and the abstract class makes the code a lot tidier. I'm not so sure about AbstractSessionStore and AbstractWindowStore as they don't offer much value, and i have a tendency to avoid unnecessarily deep inheritance (well, i'd actually prefer no impl inheritance at all!).

In Stores i think the addition of PersistentSessionWindowedFactory and PersistentTimeWindowedFactory are good. But i'm not really getting why we need PersistentWrapperFactory and the abstract classes that implement it? It seems to me that we just need a class that captures whether or not the stores are logged/cached etc. This could just be a shared class among all the factories that is created and passed to the constructor of the appropriate factory.

dguy · 2017-01-09T16:09:30Z

Something i forgot to add. In ProcessorStateManager you have removed the below block of code in flush

final ProcessorNode processorNode = stateStoreProcessorNodeMap.get(store);
                if (processorNode != null) {
                    context.setCurrentNode(processorNode);
                }

It isn't clear to me where you intend for this logic to go now?

dguy · 2017-01-09T17:16:55Z

Another thing. Point 6 you mention that we can get rid of TupleForwarder, yet i still don't see how that is possible? i.e, a user can still provide a store that is not cached - in which case we would still need it.
Perhaps you are suggesting the Store or WrappedStore takes care of the forwarding?

guozhangwang · 2017-01-09T17:46:31Z

About wappers in Stores: I think I agree. The original motivation was that we will let for example XXXStoreSupplier.build() to return XXXStore instead of not only StateStore, and hence in the DSL we will enforce Window/SessionStoreSupplier in aggregation / joins instead of just StateStoreSupplier, which actually allow users to pass a key-value store supplier that will only cause a RTE. But while implementing it I feel there maybe better ways to achieve this. So feel free to change.

About stateStoreProcessorNodeMap, my observation is that it is neither correct nor necessary: it links a store to the FIRST processor node that links to it, while multiple processors could link to this store, and it is used to set the currentProcessorNode in context when we call store.flush, but since we have already stored that information in dirtyEntry and always set/reset it in the flushing listener I feel it is not needed any more. Maybe I miss anything?

About TupleForwarder: my rationale is that, we can add some functions like wrap like I showed in the example to allow users to wrap their underlying storage engines with library-provided metering / logging / caching functionalities, AND if they don't do it, they either 1) do not really want caching or logging, which is fine; 2) they use their only logging / caching mechanism, in which case they should implement the loggingEnabled / etc functions to be false. Then in KStream/KTableXXXAggregateProcessor.process() implementation, we can decide if windowStore isInstanceOf CachingWindowStore to decide whether or not call context.foward() or not knowing that the caching store will execute this step itself; in this case we just won't do any coupling of dedupping with caching if user's customized store does that internally. And in the future, if we completely change KIP-63 to not coupling dedupping with caching but let user specifies triggers, then we can get rid of the windowStore isInstanceOf hacky logic as well.

@dguy BTW I think my refactoring is mingled with too many independent features and each of the above refactoring can be done in a single PR so we can make the reviewing process easier.

dguy · 2017-01-10T10:06:32Z

w.r.t the TupleForwarder bit - the reason it exists at the moment is to avoid duplicating the decision etc of whether context.forward() should or shouldn't be called. I don't think this is actually changed by what is suggested, but maybe i'm just not seeing it at the moment. Anyway, i'll have a look and see where we get to

guozhangwang · 2017-01-10T21:55:56Z

w.r.t the TupleForwarder bit

I think you are right. As long as we do not have the full decision on which stores should always / never be caching enabled, we cannot drop it. I guess we can only revisit it if we consider reverting the unification of caching and dedupping.

dguy · 2017-01-11T10:45:12Z

@guozhangwang - i've been trying to work with this, but it is an uphill battle. There are changes in here that definitely need a KIP, i.e, the refactoring in Stores, so that should be done separately. So i'd need to revert all those changes. Then i am having put a bunch of effort in to understand where you are heading, get it to compile, get tests to pass etc. I feel like i've been spinning my wheels and not getting anywhere.
So, i can't continue with this PR. I think the best course of action is to do this in smaller PRs. I.e, first do the StateStore hierarchy refactor. I can start that now. We need a KIP for the changes to Stores, so that is going to need to wait. There is also the TopologyBuilder etc stuff you were doing.

This PR is extracted from #2333 as an incremental fix to ease the reviewing: 1. Removed `storeToProcessorNodeMap` from ProcessorTopology since it was previously used to set the context current record, and can now be replaced with the dirty entry in the named cache. 2. Replaced `sourceStoreToSourceTopic` from ProcessorTopology with `storeToChangelogTopic` map, which includes the corresponding changelog topic name for all stores that are changelog enabled. 3. Modified `ProcessorStateManager` to rely on `sourceStoreToSourceTopic` when retrieving the changelog topic; this makes the second parameter `loggingEnabled` in `register` not needed any more, and we can deprecate the old API with a new one. 4. Also fixed a minor issue in `KStreamBuilder`: if the storeName is not provided in the `table(..)` function, do not create the underlying materialized store. Modified the unit tests to cover this case. 5. Fixed a bunch of other unit tests failures that are exposed by this refactoring, in which we are not setting the applicationId correctly when constructing the mocking processor topology. Author: Guozhang Wang <wangguoz@gmail.com> Reviewers: Damian Guy, Matthias J. Sax, Ewen Cheslack-Postava Closes #2338 from guozhangwang/KMinor-refactor-state-to-changelogtopic

asfbot · 2017-01-18T08:02:28Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/979/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-01-18T08:03:31Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/977/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-01-18T08:07:06Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/977/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-01-18T08:20:02Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/979/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-01-18T08:21:54Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/981/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-01-18T08:22:55Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/979/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-01-18T08:26:24Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/980/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-02-01T18:01:07Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/1404/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-02-01T18:01:46Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/1401/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-02-01T18:01:49Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/1401/
Test FAILed (JDK 7 and Scala 2.10).

guozhangwang · 2017-02-01T20:42:23Z

@dguy ready for review again. Some changes:

Refactored RocksDBWindowStoreTest a bit to put common variables as class variables.
Use the byte stores in RocksDBSessionStoreSupplier.

asfbot · 2017-02-01T21:22:52Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/1410/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-02-01T22:44:16Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/1407/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-02-01T22:48:27Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/1407/
Test PASSed (JDK 7 and Scala 2.10).

dguy

Overall LGTM. I left a couple of very minor nit picks.
Also, should we add tests for WrappedWindowStoreIterator and WrappedSessionStoreIterator ?

dguy · 2017-02-02T08:56:22Z

        return ByteBuffer.wrap(binaryKey).getLong(binaryKey.length - TIMESTAMP_SIZE);
    }

+    public static Window extractWindow(final byte [] binaryKey) {


Thanks for adding the test. One nit: looks like this method doesn't need to be public.

dguy · 2017-02-02T09:05:19Z

+    @Override
    public void remove(final Windowed<K> sessionKey) {
        validateStoreOpen();
+


nit: any reason for the extra line?

re One nit: looks like this method doesn't need to be public. Actually the test needs it to be package-private, and while on that, I feel we can either make it pure private and remove the corresponding test, or leave it public with the test if we speculate it to be used in the future.

I'd go with the second option if you do not have a strong opinion.

dguy · 2017-02-02T09:18:38Z

-    private static final long DEFAULT_CACHE_SIZE_BYTES = 1 * 1024 * 1024L;
-    private ThreadCache cache;
    private final Segments segments = new Segments(windowName, retentionPeriod, numSegments);
+    private final StateSerdes<Integer, String> serdes = new StateSerdes<>("", Serdes.Integer(), Serdes.String());


Nice tidy up of this test class :-)

guozhangwang · 2017-02-02T18:16:40Z

Also, should we add tests for WrappedWindowStoreIterator and WrappedSessionStoreIterator ?

I tried to add some tests, but while doing that I found all the logic is covered in corresponding unit tests of SessionKeySerde and WindowStoreUtils, so I decided to not introduce overlap coverage.

asfbot · 2017-02-02T18:45:06Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/1438/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-02-02T19:04:28Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/1435/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-02-02T20:26:56Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/1435/
Test PASSed (JDK 7 and Scala 2.10).

mjsax · 2017-02-02T22:19:47Z

    @Override
    public void remove() {
-        throw new UnsupportedOperationException("remove not supported");
+        throw new UnsupportedOperationException("remove() is not supported in DelegatingPeekingKeyValueIterator.");


Nit: "... DelegatingPeekingKeyValueIterator" -> "..." + getClass().getName()

mjsax · 2017-02-02T22:26:43Z

-    void initInternal(final ProcessorContext context) {
+    private void initInternal(final ProcessorContext context) {
        this.context = (InternalProcessorContext) context;
        this.serdes = new StateSerdes<>(underlying.name(),


Nit: remove this
(we should avoid this whenever possible)

I'd prefer to leave it here for better code reading.

mjsax · 2017-02-02T22:29:19Z

+        this.map = new TreeMap<>();
+    }
+
+    InMemoryKeyValueStore(String name) {


add final (also all other methods below)

KeyValueStore interface does not mark all the parameters as final, and I think it is safer to keep it as is. As for this constructor, will remove it.

mjsax · 2017-02-02T22:31:58Z

+
+    @Override
+    public void close() {
+        this.open = false;


Should we do map.clear() here?

mjsax · 2017-02-02T22:34:54Z

+
+        @Override
+        public void close() {
+            // do nothing


May set iter = null to ensure this iterator is not use after closing anymore?

This iterator is always wrapped in DelegatingPeekingKeyValueIterator, which already does the isOpen validation.

mjsax · 2017-02-02T23:23:42Z

+
+        @Override
+        public KeyValue<Long, byte[]> next() {
+            if (!bytesIterator.hasNext()) {


Do we need this? Should bytesIterator.next() not throw the same exception for this case?

Since the underlying iterator could be provided by users which does not validate in its own next() call, it's better doing such validation in the higher-level where the library controls.

mjsax · 2017-02-03T00:25:38Z

+
+    @Override
+    public void remove() {
+        throw new UnsupportedOperationException("remove() is not supported in WrappedWindowStoreIterator.");


mjsax · 2017-02-03T00:36:31Z

+        final KeyValueIterator<String, String> range = store.range(hello, "zooom");
        assertThat(range.next(), equalTo(KeyValue.pair(hello, world)));
        assertThat(range.next(), equalTo(KeyValue.pair(hi, there)));
        assertFalse(range.hasNext());


I am confused. If we change store.range(hello, "zooom"); should hasNext() not return true (start and end are inclusive). Thus, to me it seems the test is wrong and there must be a bug in the iterator code.

Great catch!

mjsax · 2017-02-03T00:38:55Z

        @Override
        public Long peekNextKey() {
-            throw new UnsupportedOperationException("peekNextKey not supported in stub");
+            throw new UnsupportedOperationException("peekNextKey() not supported in ReadOnlyWindowStoreStub.");


mjsax · 2017-02-03T00:43:19Z

    public void setUp() {
-        cache = new ThreadCache("testCache", DEFAULT_CACHE_SIZE_BYTES, new MockStreamsMetrics(new Metrics()));
+        baseDir = TestUtils.tempDirectory("test");
+        context = new MockProcessorContext(baseDir, Serdes.ByteArray(), Serdes.ByteArray(), recordCollector, cache);


Can't we initialize both vars directly and remove this method?

asfbot · 2017-02-03T05:36:13Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/1454/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-02-03T06:42:34Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/1457/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-02-03T06:48:15Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/1454/
Test PASSed (JDK 7 and Scala 2.10).

mjsax · 2017-02-03T06:59:28Z

        store.put(hi, there);
        store.put("zooom", "home");
-        final KeyValueIterator<String, String> range = store.range(hello, "zooom");
+        final KeyValueIterator<String, String> range = store.range(hello, hi);


I am still puzzled why the test did not fail with "zooom" as end range. Can you explain? \cc @dguy WDYT?

Because in the old code of InMemoryKeyValueStore, our range function is implemented as subMap(from, true, to, false) while I changed it to (from, true, to, true) in the latest commit.

Ack. Overlooked this change :)

mjsax

LGTM

junrao · 2017-02-03T18:15:50Z

Thanks for the patch. LGTM

guozhangwang · 2017-02-03T19:13:09Z

Merged to trunk.

This PR is extracted from apache#2333 as an incremental fix to ease the reviewing: 1. Removed `storeToProcessorNodeMap` from ProcessorTopology since it was previously used to set the context current record, and can now be replaced with the dirty entry in the named cache. 2. Replaced `sourceStoreToSourceTopic` from ProcessorTopology with `storeToChangelogTopic` map, which includes the corresponding changelog topic name for all stores that are changelog enabled. 3. Modified `ProcessorStateManager` to rely on `sourceStoreToSourceTopic` when retrieving the changelog topic; this makes the second parameter `loggingEnabled` in `register` not needed any more, and we can deprecate the old API with a new one. 4. Also fixed a minor issue in `KStreamBuilder`: if the storeName is not provided in the `table(..)` function, do not create the underlying materialized store. Modified the unit tests to cover this case. 5. Fixed a bunch of other unit tests failures that are exposed by this refactoring, in which we are not setting the applicationId correctly when constructing the mocking processor topology. Author: Guozhang Wang <wangguoz@gmail.com> Reviewers: Damian Guy, Matthias J. Sax, Ewen Cheslack-Postava Closes apache#2338 from guozhangwang/KMinor-refactor-state-to-changelogtopic

This is a refactoring follow-up of apache#2166. Main refactoring changes: 1. Extract `InMemoryKeyValueStore` out of `InMemoryKeyValueStoreSupplier` and remove its duplicates in test package. 2. Add two abstract classes `AbstractKeyValueIterator` and `AbstractKeyValueStore` to collapse common functional logics. 3. Added specialized `BytesXXStore` to accommodate cases where key value types are Bytes / byte[] so that we can save calling the dummy serdes. 4. Make the key type in `ThreadCache` from byte[] to Bytes, as SessionStore / WindowStore's result serialized bytes are in the form of Bytes anyways, so that we can save unnecessary `Bytes.get()` and `Bytes.wrap(bytes)`. Each of these should arguably be a separate PR and I apologize for the mess, this is because this branch was extracted from a rather large diff that has multiple refactoring mingled together and dguy and myself have already put lots of efforts to break it down to a few separate PRs, and this is the only left-over work. Such PR won't happen in the future. Ping dguy enothereska mjsax for reviews Author: Guozhang Wang <wangguoz@gmail.com> Reviewers: Damian Guy, Matthias J. Sax, Jun Rao Closes apache#2333 from guozhangwang/K3452-followup-state-store-refactor

This PR is extracted from apache/kafka#2333 as an incremental fix to ease the reviewing: 1. Removed `storeToProcessorNodeMap` from ProcessorTopology since it was previously used to set the context current record, and can now be replaced with the dirty entry in the named cache. 2. Replaced `sourceStoreToSourceTopic` from ProcessorTopology with `storeToChangelogTopic` map, which includes the corresponding changelog topic name for all stores that are changelog enabled. 3. Modified `ProcessorStateManager` to rely on `sourceStoreToSourceTopic` when retrieving the changelog topic; this makes the second parameter `loggingEnabled` in `register` not needed any more, and we can deprecate the old API with a new one. 4. Also fixed a minor issue in `KStreamBuilder`: if the storeName is not provided in the `table(..)` function, do not create the underlying materialized store. Modified the unit tests to cover this case. 5. Fixed a bunch of other unit tests failures that are exposed by this refactoring, in which we are not setting the applicationId correctly when constructing the mocking processor topology. Author: Guozhang Wang <wangguoz@gmail.com> Reviewers: Damian Guy, Matthias J. Sax, Ewen Cheslack-Postava Closes #2338 from guozhangwang/KMinor-refactor-state-to-changelogtopic

a first pass of the refactoring

575226c

refactor processor topology a bit to let processor manager be aware o…

1cbfb58

…f logged stores beforehand

guozhangwang changed the title ~~KAFKA-3452 Follow-up: Refactoring StateStore hierarchies~~ [WIP] KAFKA-3452 Follow-up: Refactoring StateStore hierarchies Jan 9, 2017

dguy reviewed Jan 9, 2017

View reviewed changes

guozhangwang mentioned this pull request Jan 10, 2017

MINOR: remove unnecessary store info from TopologyBuilder #2338

Closed

rebase on apache trunk

9330c9a

minor fixes

8de39d6

additional fixes

86e4bed

guozhangwang added 2 commits February 1, 2017 09:58

github comments v3

dcb9e60

extracted wrapper iterators

ef9e66b

address github comments, v4

f1b52ef

dguy reviewed Feb 2, 2017

View reviewed changes

minor fixes

81e4641

mjsax reviewed Feb 3, 2017

View reviewed changes

github comments

b482f4c

mjsax reviewed Feb 3, 2017

View reviewed changes

mjsax approved these changes Feb 3, 2017

View reviewed changes

asfgit closed this in 7ebc5da Feb 3, 2017

guozhangwang deleted the K3452-followup-state-store-refactor branch July 15, 2017 22:07

Conversation

guozhangwang commented Jan 9, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

asfbot commented Jan 9, 2017

Uh oh!

asfbot commented Jan 9, 2017

Uh oh!

asfbot commented Jan 9, 2017

Uh oh!

asfbot commented Jan 9, 2017

Uh oh!

asfbot commented Jan 9, 2017

Uh oh!

asfbot commented Jan 9, 2017

Uh oh!

guozhangwang commented Jan 9, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dguy commented Jan 9, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

dguy commented Jan 9, 2017

Uh oh!

dguy commented Jan 9, 2017

Uh oh!

guozhangwang commented Jan 9, 2017

Uh oh!

dguy commented Jan 10, 2017

Uh oh!

guozhangwang commented Jan 10, 2017

Uh oh!

dguy commented Jan 11, 2017

Uh oh!

asfbot commented Jan 18, 2017

Uh oh!

asfbot commented Jan 18, 2017

Uh oh!

asfbot commented Jan 18, 2017

Uh oh!

asfbot commented Jan 18, 2017

Uh oh!

asfbot commented Jan 18, 2017

Uh oh!

asfbot commented Jan 18, 2017

Uh oh!

asfbot commented Jan 18, 2017

Uh oh!

asfbot commented Feb 1, 2017

Uh oh!

asfbot commented Feb 1, 2017

Uh oh!

asfbot commented Feb 1, 2017

Uh oh!

guozhangwang commented Feb 1, 2017

Uh oh!

asfbot commented Feb 1, 2017

Uh oh!

asfbot commented Feb 1, 2017

Uh oh!

asfbot commented Feb 1, 2017

Uh oh!

dguy left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

guozhangwang commented Feb 2, 2017

Uh oh!

guozhangwang commented Jan 9, 2017 •

edited

Loading

dguy commented Jan 9, 2017 •

edited

Loading

mjsax Feb 2, 2017 •

edited

Loading