Author: Matthias J. Sax <matthias@confluent.io>
Reviewers: Damian Guy <damian.guy@gmail.com>, Guozhang Wang <wangguoz@gmail.com>
Closes#1940 from mjsax/hotfix-resetTool
restoreConsumer.assign(..) in removeStandbyTasks was logging an (ignorable) exception due to the restoreConsumer being closed. Moved the restoreConsumer.assign(..) to shutdownTasksAndState as this is done prior to the closing of consumers.
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Closes#1986 from dguy/hotfix-assign
Changed the ordering in `StreamThread.shutdown`
1. commitAll (we need to commit so that any cached data is flushed through the topology)
2. close all tasks
3. producer.flush() - so any records produced during close are flushed and we have offsets for them
4. close all state managers
5. close producers/consumers
6. remove the tasks
Also in `onPartitionsRevoked`
1. commitAll
2. close all tasks
3. producer.flush
4. close all state managers
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Eno Thereska, Guozhang Wang
Closes#1970 from dguy/kafka-4253
pass through user-defined consumer and producer properties from StreamsConfig to ConsumerConfig and ProducerConfig.
For example, consumer interceptor support, i.e, consumer.interceptor.classes=SomeClass
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Closes#1943 from dguy/interceptor
Author: Matthias J. Sax <matthias@confluent.io>
Reviewers: Guozhang Wang <wangguoz@gmail.com>, Eno Thereska <eno.thereska@gmail.com>, Ismael Juma <ismael@juma.me.uk>
Closes#1938 from mjsax/kafka-4058-resetIntegartionTest
Change the creation of the directories, in the StateDirectory constructor, to use mkdirs so any parents get created. Throw an exception if the directory doesn't exist and couldn't be created
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Michael G. Noll, Eno Thereska, Guozhang Wang
Closes#1942 from dguy/kafka-4233
Author: Guozhang Wang <wangguoz@gmail.com>
Reviewers: Eno Thereska, Damian Guy
Closes#1929 from guozhangwang/KMinor-check-zero-num-partitions and squashes the following commits:
c8edc2d [Guozhang Wang] fix unit test
3127f9c [Guozhang Wang] do not call partitioner if num partitions is non-positive
Turning off auto topic creation in the EmbeddedKafkaCluster used by Streams as it can cause race conditions that lead to build hangs.
Fixed the couple of tests that needed to have some topics manually created
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Eno Thereska <eno.thereska@gmail.com>, Ismael Juma <ismael@juma.me.uk>
Closes#1941 from dguy/disable-auto-topic-create
Keep track of open Rocks DB iterators. When a store is closed, close all open iterators.
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Closes#1917 from dguy/kafka-4223
KafkaExceptions currently thrown from within StreamThread/StreamTask currently bubble up without any additional context. This makes it hard to figure out where something went wrong, i.e, which topic had the serialization exception etc
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Closes#1819 from dguy/kafka-3708 and squashes the following commits:
d6feaa8 [Damian Guy] address comments
15b89e7 [Damian Guy] merge trunk
6b8a8af [Damian Guy] catch exceptions in various places and throw more informative versions
c86eeda [Damian Guy] fix conflicts
8f37e2c [Damian Guy] add some context to exceptions
Adding the test so we know that the State Stores with logging disabled or without a topic don't throw any exceptions.
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Closes#1916 from dguy/state-store-logging-disabled
If some StreamsMetadataState methods are called before the onChange method is called a NullPointerException was being thrown. Added null check for cluster in isInitialized method
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Closes#1920 from dguy/fix-npe-streamsmetadata
The original commit interval of 30 seconds might be too large in some cases, e.g., when the verifier finishes before those 30 seconds have elapsed.
Author: Eno Thereska <eno.thereska@gmail.com>
Reviewers: Damian Guy, Guozhang Wang
Closes#1899 from enothereska/hotfix-smoke-test-commit-interval
Remove isValidCleanupPolicy and related fields as they are never used.
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Eno Thereska, Guozhang Wang
Closes#1888 from dguy/minor-remove-unused
missing javadoc on public API method PersistenKeyValueFactory.enableCaching
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Eno Thereska, Guozhang Wang
Closes#1891 from dguy/minor-java-doc
Author: Eno Thereska <eno.thereska@gmail.com>
Reviewers: Damian Guy <damian.guy@gmail.com>, Ismael Juma <ismael@juma.me.uk>
Closes#1887 from enothereska/hotfix-metadata-unavailable
The contribution is my original work and I license the work to the project under the project's open source license.
guozhangwang
Author: Elias Levy <fearsome.lucidity@gmail.com>
Reviewers: Damian Guy, Matthias J. Sax, Guozhang Wang
Closes#1846 from eliaslevy/KAFKA-4153
During rebalance operations the Cluster object gets set to Cluster.empty(). This can result in NPEs when doing certain operation on StreamsMetadataState. This should throw a StreamsException if the Cluster is empty as it is not yet (re-)initialized
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Eno Thereska, Guozhang Wang
Closes#1845 from dguy/streams-meta-hotfix
standby tasks should be assigned per consumer not per process
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Eno Thereska, Guozhang Wang
Closes#1862 from dguy/kafka-4175
The logic in `verifyCanGetByKey` was incorrect. It was
```
windowState.size() < keys.length &&
countState.size() < keys.length &&
System.currentTimeMillis() < timeout
```
but should be:
```
(windowState.size() < keys.length || countState.size() < keys.length) && System.currentTimeMillis() < timeout
```
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Closes#1879 from dguy/minor-fix-test
This PR implements KIP-78:Cluster Identifiers [(link)](https://cwiki.apache.org/confluence/display/KAFKA/KIP-78%3A+Cluster+Id#KIP-78:ClusterId-Overview) and includes the following changes:
1. Changes to broker code
- generate cluster id and store it in Zookeeper
- update protocol to add cluster id to metadata request and response
- add ClusterResourceListener interface, ClusterResource class and ClusterMetadataListeners utility class
- send ClusterResource events to the metric reporters
2. Changes to client code
- update Cluster and Metadata code to support cluster id
- update clients for sending ClusterResource events to interceptors, (de)serializers and metric reporters
3. Integration tests for interceptors, (de)serializers and metric reporters for clients and for protocol changes and metric reporters for broker.
4. System tests for upgrading from previous versions.
Author: Sumit Arrawatia <sumit.arrawatia@gmail.com>
Author: Ismael Juma <ismael@juma.me.uk>
Reviewers: Jun Rao <junrao@gmail.com>, Ismael Juma <ismael@juma.me.uk>
Closes#1830 from arrawatia/kip-78
This is joint work between dguy and enothereska. The work implements KIP-63. Overview of main changes:
- New byte-based cache that acts as a buffer for any persistent store and for forwarding changes downstream.
- Forwarding record path changes: previously a record in a task completed end-to-end. Now it may be buffered in a processor node while other records complete in the task.
- Cleanup and state stores and decoupling of cache from state store and forwarding.
- More than 80 new unit and integration tests.
Author: Damian Guy <damian.guy@gmail.com>
Author: Eno Thereska <eno.thereska@gmail.com>
Reviewers: Matthias J. Sax, Guozhang Wang
Closes#1752 from enothereska/KAFKA-3776-poc
Fix for bug outlined in KAFKA-4131
Author: bbejeck <bbejeck@gmail.com>
Reviewers: Damian Guy, Guozhang Wang
Closes#1843 from bbejeck/KAFKA-4131_mulitple_regex_consumers_cause_npe
Set the NUM_STREAM_THREADS_CONFIG = 1 in SmokeTestClient as we get locking issues when we have NUM_STREAM_THREADS_CONFIG > 1 and we have Standby Tasks, i.e., replicas. This is because the Standby Tasks can be assigned to the same KafkaStreams instance as the active task, hence the directory is locked
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Eno Thereska, Guozhang Wang
Closes#1861 from dguy/fix-smoketest
A couple of the tests may transiently fail in QueryableStateIntegrationTest as they are not catching InvalidStateStoreException. This exception is expected during rebalance.
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Eno Thereska, Guozhang Wang
Closes#1840 from dguy/minor-fix
The verification in verifyGreaterOrEqual was incorrect. It was failing when a new key was found.
Set the TimeWindow to a large value so all windowed results fall in a single window
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Closes#1833 from dguy/minor-test-fix
changelogs of window stores now configure cleanup.policy=compact,delete with retention.ms set to window maintainMs + StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG
StoreChangeLogger produces messages with context.timestamp().
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Eno Thereska, Guozhang Wang
Closes#1792 from dguy/kafka-3595
Mark the store as open after the DB has been restored from the changelog.
Only add the store to the map in ProcessorStateManager post restore.
Make RocksDBWindowStore.Segment override openDB(..) as it needs to mark the Segment as open.
Throw InvalidStateStoreException if any stores in a KafkaStreams instance are not available.
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Eno Thereska, Guozhang Wang
Closes#1824 from dguy/kafka-4123
- use AdminTool to check for active consumer group
Author: Matthias J. Sax <matthias@confluent.io>
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Closes#1767 from mjsax/kafka-4058-trunk
If the thread or process is not the coordinator the Cluster instance in StreamPartitionAssignor will always be null. This builds an instance of the Cluster with the metadata associated with the Assignment
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Closes#1804 from dguy/kafka-4104
There's a minor bug in ProcessorTopologyTestDriver that prevents it from working with a topology that contains multiple sources. The bug is that ```consumer.assign()``` is called while looping through all the source topics, but, consumer.assign resets the state of the MockConsumer to only consume from the topics passed in. This patch fixes the issue by calling consumer.assign once with all the TopicPartition instances. Unit test (testDrivingSimpleMultiSourceTopology) included.
This contribution is my original work and I license the work to the project under the project's open source license.
Author: Mathieu Fenniak <mathieu.fenniak@replicon.com>
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Closes#1782 from mfenniak/ProcessorTopologyTestDriver-multiple-source-bugfix
Author: Guozhang Wang <wangguoz@gmail.com>
Reviewers: Damian Guy <damian.guy@gmail.com>, Ewen Cheslack-Postava <ewen@confluent.io>
Closes#1736 from guozhangwang/Kminor-topology-applicationID
Author: Guozhang Wang <wangguoz@gmail.com>
Reviewers: Ismael Juma <ismael@juma.me.uk>, Ewen Cheslack-Postava <me@ewencp.org>
Closes#1746 from guozhangwang/K4049-RegexSourceIntegrationTest-failure
Author: Guozhang Wang <wangguoz@gmail.com>
Reviewers: Damian Guy <damian.guy@gmail.com>, Matthias J. Sax <matthias@confluent.io>, Michael G. Noll <michael@confluent.io>, Greg Fodor <gfodor@gmail.com>, Ismael Juma <ismael@juma.me.uk>, Ewen Cheslack-Postava <ewen@confluent.io>
Closes#1530 from guozhangwang/K3769-per-thread-metrics
guozhangwang enothereska mjsax miguno please take a look. A few things that need to be clarified
1. I've added StreamsConfig.USER_ENDPOINT_CONFIG, but should we have separate configs for host and port or is this one config ok?
2. `HostState` in the KIP has a byte[] field - not sure why and what it would be populated with
3. I've changed the API to return `List<KafkaStreamsInstance>` as opposed to `Map<HostInfo, Set<TaskMetadata>>` as i find this far more intuitive to work with.
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Matthias J. Sax, Michael G. Noll, Eno Thereska, Guozhang Wang
Closes#1576 from dguy/kafka-3914v2
Added non null checks to parameters supplied via the DSL and `TopologyBuilder`
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Edward Ribeiro <edward.ribeiro@gmail.com>, Guozhang Wang <wangguoz@gmail.com>
Closes#1711 from dguy/kafka-3936
It previously hardcoded it.
Author: Ismael Juma <ismael@juma.me.uk>
Reviewers: Grant Henke <granthenke@gmail.com>, Jason Gustafson <jason@confluent.io>, Ewen Cheslack-Postava <ewen@confluent.io>
Closes#1613 from ijuma/kafka-3954-consumer-internal-topics-from-broker
I affirm that the contribution is my original work and that I license the work to the project under the project's open source license.
This cleans up misbehaviour that was introduce while fixing KAFKA-3817. It is impossible for a non-count aggregate to be build, when the addition happens before the removal. IMHO making sure that these details are correct is very important.
This PR has local test errors. It somehow fails the ResetIntegrationTest. It doesn't quite appear to me why but it looks like this PR breaks it, especially because the error appears with the ordering of the events. Still I am unable to find where I could have broken it. Maybe not seems to fail on trunk aswell.
Author: jfilipiak <Jan.Filipiak@trivago.com>
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Closes#1705 from Kaiserchen/KAFKA-3817-preserve-order-for-aggreagators
Rename StateStoreProvider.getStores(...) to StateStoreProvider.stores(...) as this is consistent with the naming of other 'getters' in the public API.
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Guozhang Wang <wangguoz@gmail.com>, Ismael Juma <ismael@juma.me.uk>
Closes#1699 from dguy/minor-method-rename
Matthias J. Sax
Damian Guy
Guozhang Wang
Eno Thereska
Elias Levy
Dan Norwood
Sumit Arrawatia
Bill Bejeck
Michael G. Noll
Mathieu Fenniak
Ismael Juma
Jan Filipiak