1. In StreamThread, always use subscribe(Pattern, ..) function in order to avoid sending MetadataRequest with specific topic names and cause brokers to possibly auto-create subscribed topics; the pattern is generated as "topic-1|topic-2..|topic-n".
2. In ConsumerCoordinator, let the leader to refresh its metadata if the generated assignment contains some topics that is not contained in the subscribed topics; also in SubscriptionState, modified the verification for regex subscription to against the regex pattern instead of the matched topics since the returned assignment may contain some topics not yet created when joining the group but existed after the rebalance; also modified some unit tests in `KafkaConsumerTest` to accommodate the above changes.
3. Minor cleanup: changed String[] to List<String> to avoid overloaded functions.
4. Minor cleanup: enforced strong typing in SinkNodeFactory and removed unnecessary unchecked tags.
5. Minor cleanup: augmented unit test error message and fixed a potential transient failure in KafkaStreamTest.
Author: Guozhang Wang <wangguoz@gmail.com>
Reviewers: Damian Guy <damian.guy@gmail.com>, Matthias J. Sax <matthias@confluent.io>, Jason Gustafson <jason@confluent.io>
Closes#2379 from guozhangwang/K4633-regex-pattern
Variance changes introduced in KIP-100 cause compilation failures with lambda expression in Java 8. To my knowledge this only affects the following method
`KStreams.transform(TransformerSupplier<...>, String...)`
prior to the changes it was possible to write:
`streams.transform(MyTransformer::new)`
where `MyTransformer` extends `Transformer`
After the changes the Java compiler is unable to infer correct return types for the lambda expressions. This change fixed this by reverting to invariant return types for transformer suppliers.
please cherry-pick into 0.10.2.x
Author: Xavier Léauté <xavier@confluent.io>
Reviewers: Ismael Juma, Damian Guy, Guozhang Wang
Closes#2402 from xvrl/lambdas-oh-my
ZK removed reveal a bug in `StreamPartitionAssigner` but did not fix it properly. This is a follow up bug fix.
Issue:
- If topic metadata is missing, `StreamPartitionAssigner` should not create any affected tasks that consume topics with missing metadata.
- Depending downstream tasks should not be create either.
- For tasks that are not created, no store changelog topics (if any) should get created
- For tasks that write output to not-yet existing internal repartitioning topics, those repartitioning topics should not get created
Author: Matthias J. Sax <matthias@confluent.io>
Reviewers: Damian Guy, Guozhang Wang
Closes#2404 from mjsax/kafka-4060-zk-test-follow-up
This log message tends to be extremely verbose when state stores are being restored
Author: Xavier Léauté <xavier@confluent.io>
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Closes#2412 from xvrl/reduce-verbosity
Re-branched the trunk and applied the changes to the new branch to simplify commit log.
Author: Hojjat Jafarpour <hojjat@Hojjat-Jafarpours-MBP.local>
Reviewers: Ismael Juma, Damian Guy, Eno Thereska, Guozhang Wang
Closes#2389 from hjafarpour/KAFKA-4060-Remove-ZkClient-dependency-in-Kafka-Streams-followup-from-trunk
Address Ismael's comments upon merging
This class doesn't need to override this method as it is handled appropriately by the super class
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Closes#2397 from dguy/hotfix-npe-state-store
interface for `Processor` in comments incorrectly had `transform` rather than `process`.
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Michael G. Noll, Ismael Juma <ismael@juma.me.uk>
Closes#2396 from dguy/minor-javadoc
In RocksDBStore, options / wOptions / fOptions are constructed in the constructor, which needs to be dismissed in the close() call; however in some tests, the generated topology is not initialized at all, and hence the corresponding state stores are supposed to not be able to be closed as well since their `init` function is not called. This could cause the above option objects to be not released.
This is fixed in this patch to move the logic out of constructor and inside `init` functions, so that no RocksDB objects will be created in the constructor only. Also some minor cleanups:
1. In KStreamTestDriver.close(), we lost the logic to close the state stores but only call `flush`; it is now changed back to call both.
2. Moved the forwarding logic from KStreamTestDriver to MockProcessorContext to remove the mutual dependency: these functions should really be in ProcessorContext, not the test driver.
Author: Guozhang Wang <wangguoz@gmail.com>
Reviewers: Damian Guy <damian.guy@gmail.com>, Matthias J. Sax <matthias@confluent.io>, Jason Gustafson <jason@confluent.io>
Closes#2381 from guozhangwang/K3502-pure-virtual-function-unit-tests
Author: Matthias J. Sax <matthias@confluent.io>
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Closes#2391 from mjsax/kafka-4060-zk-follow-up-system-tests
This is a follow up of https://github.com/apache/kafka/pull/2166 - refactoring the store hierarchies as requested
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Closes#2360 from dguy/state-store-refactor
After debugging this i can see the times that it fails there is a race between when the topic is actually created/ready on the broker and when the assignment happens. When it fails `StreamPartitionAssignor.assign(..)` gets called with a `Cluster` with no topics. Hence the test hangs as no tasks get assigned. To fix this I added a `waitForTopics` method to `EmbeddedKafkaCluster`. This will wait until the topics have been created.
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Matthias J. Sax, Guozhang Wang
Closes#2371 from dguy/integration-test-fix
Remove applicationId parameter as it is no longer used.
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Closes#2385 from dguy/minor-remove-unused-param
This PR is extracted from https://github.com/apache/kafka/pull/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
Author: Ismael Juma <ismael@juma.me.uk>
Reviewers: Gwen Shapira <cshapi@gmail.com>, Jason Gustafson <jason@confluent.io>
Closes#2354 from ijuma/kafka-4565-separation-of-internal-and-external-traffic
and remove some unnecessary SuppressWarnings annotations
Author: Xavier Léauté <xavier@confluent.io>
Reviewers: Ismael Juma, Guozhang Wang
Closes#2363 from xvrl/kip-100-followup
Add Global Tables to KafkaStreams. Global Tables are fully replicated once-per instance of KafkaStreams. A single thread is used to update them. They can be used to join with KStreams, KTables, and other GlobalKTables. When participating in a join a GlobalKTable is only ever used to perform a lookup, i.e., it will never cause data to be forwarded to downstream processor nodes.
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Matthias J. Sax, Eno Thereska, Guozhang Wang
Closes#2244 from dguy/global-tables
- TimeWindows represent half-open time intervals while SessionWindows represent closed time intervals
Author: Matthias J. Sax <matthias@confluent.io>
Reviewers: Damian Guy <damian.guy@gmail.com>, Guozhang Wang <wangguoz@gmail.com>
Closes#2342 from mjsax/kafka-3452-session-window-follow-up
mjsax
Here's my first pass at finer grained auto offset reset strategies.
I've left TODO comments about whether we want to consider adding this to `KGroupedTable.aggregate` and `KStreamImpl` when re-partitioning a source.
Author: bbejeck <bbejeck@gmail.com>
Reviewers: Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <wangguoz@gmail.com>
Closes#2007 from bbejeck/KAFKA-4114_allow_different_offset_reset_strategies
Kafka Streams: add granular metrics per node and per task, also expose ability to register non latency metrics in StreamsMetrics
Also added different recording levels to Metrics.
This is joint contribution from Eno Thereska and Aarti Gupta.
from https://github.com/apache/kafka/pull/1362#issuecomment-218326690-------
We can consider adding metrics for process / punctuate / commit rate at the granularity of each processor node in addition to the global rate mentioned above. This is very helpful in debugging.
We can consider adding rate / total cumulated metrics for context.forward indicating how many records were forwarded downstream from this processor node as well. This is helpful in debugging.
We can consider adding metrics for each stream partition timestamp.
This is helpful in debugging.
## Besides the latency metrics, we can also add throughput latency in terms of source records consumed.
More discussions here https://issues.apache.org/jira/browse/KAFKA-3715, KIP-104, KIP-105
Author: Eno Thereska <eno@confluent.io>
Author: Aarti Gupta <aartiguptaa@gmail.com>
Reviewers: Greg Fodor, Ismael Juma, Damian Guy, Guozhang Wang
Closes#1446 from aartigupta/trunk
The client should send older versions of requests to the broker if necessary.
Author: Colin P. Mccabe <cmccabe@confluent.io>
Reviewers: Jason Gustafson <jason@confluent.io>, Jun Rao <junrao@gmail.com>, Ismael Juma <ismael@juma.me.uk>
Closes#2264 from cmccabe/KAFKA-4507
dguy guozhangwang This is a new PR for KAFKA-4060.
Author: Hojjat Jafarpour <hojjat@Hojjat-Jafarpours-MBP.local>
Author: Hojjat Jafarpour <hojjat@HojjatJpoursMBP.attlocal.net>
Reviewers: Damian Guy, Matthias J. Sax, Isamel Juma, Guozhang Wang
Closes#1884 from hjafarpour/KAFKA-4060-Remove-ZkClient-dependency-in-Kafka-Streams-new
Make appropriate methods contravariant in key and value types.
Author: Xavier Léauté <xavier@confluent.io>
Reviewers: Damian Guy <damian.guy@gmail.com>, Guozhang Wang <wangguoz@gmail.com>
Closes#2205 from xvrl/streams-contravariance
Rename `SessionStore.findSessionsToMerge` to `findSessions`
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Closes#2339 from dguy/minor-findsession-rename
Closed the last PR I made for this because I accidentally borked it with my other PR. Small error; I figure this is from copy-pasting the above doc
Author: Nikki Thean <nthean@etsy.com>
Reviewers: Matthias J. Sax <matthias@confluent.io>, Ewen Cheslack-Postava <ewen@confluent.io>
Closes#2336 from nixsticks/trunk
When creating the source node in TopologyBuilder, we need to decorate its input topics if they are inner (i.e. repartition) topics with the prefix.
Also did some minor cleanup in the printing function for better visualization in debugging.
Author: Guozhang Wang <wangguoz@gmail.com>
Reviewers: Eno Thereska, Damian Guy, Eno Thereska, Jun Rao
Closes#2320 from guozhangwang/KMinor-source-topic-fix
Remove use of TestTimestampExtractor as it causes the logs to roll and segments get deleted.
Remove the wcnt example as it is dependent on the TestTimestampExtractor - windowed counting is covered elsewhere.
Change all aggregate operations to use TimeWindow as use of UnlimitedWindow was causing logs to roll and segments being deleted.
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Matthias J. Sax, Guozhang Wang, Eno Thereska
Closes#2319 from dguy/smoke-test
Add support for SessionWindows based on design detailed in https://cwiki.apache.org/confluence/display/KAFKA/KIP-94+Session+Windows.
This includes refactoring of the RocksDBWindowStore such that functionality common with the RocksDBSessionStore isn't duplicated.
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Eno Thereska <eno.thereska@gmail.com>, Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <wangguoz@gmail.com>
Closes#2166 from dguy/kafka-3452-session-merge
Jason recently cleaned things up significantly by consolidating the Message/Record classes
into the common Java code in the clients module. While reviewing that, I noticed a few things
that could be improved a little more. To make reviewing easier, there will be multiple PRs.
Author: Ismael Juma <ismael@juma.me.uk>
Reviewers: Ewen Cheslack-Postava <me@ewencp.org>, Jason Gustafson <jason@confluent.io>
Closes#2271 from ijuma/records-minor-fixes
In `shutdownTasksAndState` and `suspendTasksAndState` we commit offsets BEFORE we flush any state. This is wrong as if an exception occurs during a flush, we may violate the at-least-once guarantees, that is we would have committed some offsets but NOT sent the processed data on to other Sinks.
Also during suspend and shutdown, we should try and complete all tasks even when exceptions occur. We should just keep track of the exception and rethrow it at the end if necessary. This helps with ensuring that StateStores etc are closed.
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Closes#2281 from dguy/kafka-4561
This was changed in b58b6a1bef and caused the `ReplicaVerificationToolTest.test_replica_lags`
system test to start failing.
I also added a unit test and a couple of other minor clean-ups.
Author: Ismael Juma <ismael@juma.me.uk>
Reviewers: Jason Gustafson <jason@confluent.io>
Closes#2280 from ijuma/kafka-4554-fix-replica-buffer-verify-checksum
During `onPartitionsAssigned` first close, and remove, any suspended `StandbyTasks` that are no longer assigned to this consumer.
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Closes#2266 from dguy/kafka-4540
The `RecordCollectorImpl` currently drops messages on the floor if an exception is non-null in the producer callback. This will result in message loss and violates at-least-once processing.
Rather than just log an error in the callback, save the exception in a field. On subsequent calls to `send`, `flush`, `close`, first check for the existence of an exception and throw a `StreamsException` if it is non-null. Also, in the callback, if an exception has already occurred, the `offsets` map should not be updated.
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Closes#2249 from dguy/kafka-4473
partitionsByHostState and metadataWithInternalTopics need to be updated on each call to onAssignment() otherwise they contain invalid/stale metadata.
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Matthias J. Sax, Guozhang Wang
Closes#2256 from dguy/4534
Tasks that don't have any `StateStore`s wont have a `StandbyTask`, so `createStandbyTask` can return `null`. We need to check for this in `StandbyTaskCreator.createTask(...)`
Also, the checkpointed offsets for `StandbyTask`s are never loaded.
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Eno Thereska, Matthias J. Sax, Guozhang Wang
Closes#2255 from dguy/kafka-4539
Guozhang Wang
Matthias J. Sax
Xavier Léauté
Hojjat Jafarpour
Damian Guy
Ismael Juma
Xi Hu
Eno Thereska
Bill Bejeck
Eno Thereska
Colin P. Mccabe
Prabhat Kashyap
Nikki Thean