Currently, IQv2 forwards all queries to the underlying store. We add this bypass to allow handling of key queries in the cache. If a key exists in the cache, it will get answered from there.
As part of this PR, we realized we need access to the position of the underlying stores. So, I added the method getPosition to the public API and ensured all state stores implement it. Only the "leaf" stores (Rocks*, InMemory*) have an actual position, all wrapping stores access their wrapped store's position.
Reviewers: Patrick Stuedi <pstuedi@apache.org>, John Roesler <vvcephei@apache.org>
There is a brief window between when the store is registered and when
it is initialized when it might handle a query, but there is no context.
We treat this condition just like a store that hasn't caught up to the
desired position yet.
Reviewers: Guozhang Wang <guozhang@apache.org>, Matthias J. Sax <mjsax@apache.org>, A. Sophie Blee-Goldman <ableegoldman@apache.org>, Patrick Stuedi <pstuedi@apache.org>
Followup to #11600 to invoke the streams exception handler on the MissingSourceTopicException, without killing/replacing the thread
Reviewers: Guozhang Wang <guozhang@confluent.io>, Bruno Cadonna <cadonna@confluent.io>
Fixes some issues with the NamedTopology version of the IQ methods that accept a topologyName argument, and adds tests for all.
Reviewers: Guozhang Wang <guozhang@confluent.io>, Walker Carlson <wcarlson@confluent.io>
Avoid throwing a MissingSourceTopicException inside the #assign method when named topologies are used, and just remove those topologies which are missing any of their input topics from the assignment.
Reviewers: Guozhang Wang <guozhang@confluent.io>, Walker Carlson <wcarlson@confluent.io>, Bruno Cadonna <cadonna@confluent.io>
During some recent reviews, @mjsax pointed out that StateStore layers
are constructed differently the stores are added via the PAPI vs. the DSL.
This PR adds PAPI construction for Window and Session stores to the
IQv2StoreIntegrationTest so that we can ensure IQv2 works on every
possible state store.
Reviewer: Guozhang Wang <guozhang@apache.org>
During some recent reviews, @mjsax pointed out that StateStore layers
are constructed differently the stores are added via the PAPI vs. the DSL.
This PR adds KeyValueStore PAPI construction to the
IQv2StoreIntegrationTest so that we can ensure IQv2 works on every
possible state store.
Reviewers: Patrick Stuedi <pstuedi@apache.org>, Guozhang Wang <guozhang@apache.org>
Implement the KeyQuery as proposed in KIP-796
Reviewers: Vicky Papavasileiou <vpapavasileiou@confluent.io>, Matthias J. Sax <mjsax@apache.org>, Guozhang Wang <guozhang@apache.org>
When this was made I didn't expect deleteOffsetsResult to be set if an exception was thrown. But it is and to retry we need to reset it to null. Changing the KafkaStreamsNamedTopologyWrapper for remove topology when resetting offsets to retry upon GroupSubscribedToTopicException and swallow/complete upon GroupIdNotFoundException
Reviewers: Anna Sophie Blee-Goldman <ableegoldman@ache.>
Update an unclear log message and method name in TopologyMetadata
Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>, Guozhang Wang <guozhang@confluent.io>, Luke Chen <showuon@confluent.io>
* Fill in the Position response in the IQv2 result.
* Enforce PositionBound in IQv2 queries.
* Update integration testing approach to leverage consistent queries.
Reviewers: Patrick Stuedi <pstuedi@apache.org>, Vicky Papavasileiou <vpapavasileiou@confluent.io>, Guozhang Wang <guozhang@apache.org>
Fix for one of the causes of failure in the NamedTopologyIntegrationTest: org.apache.kafka.streams.errors.StreamsException: java.lang.IllegalStateException: Must initialize prevActiveTasks from ownedPartitions before initializing remaining tasks.
This exception could occur if a member sent in a subscription where all of its ownedPartitions were from a named topology that is no longer recognized by the group leader, eg because it was just removed from the client. We should filter each ClientState based on the current topology only so the assignor only processes the partitions/tasks it can identify. The member with the out-of-date tasks will eventually clean them up when the #removeNamedTopology API is invoked on them
Reviewers: Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <guozhang@confluent.io>, Walker Carlson <wcarlson@confluent.io>
Use the name specified via consumed parameter in InternalStreamsBuilder#addGlobalStore method for initializing the source name and processor name. If not specified, the names are generated.
Reviewers: Luke Chen <showuon@gmail.com>, Bill Bejeck <bbejeck@apache.org>
In the new NamedTopology API being worked on, state store names and their uniqueness requirement is going to be scoped only to the owning topology, rather than to the entire app. In other words, two different named topologies can have different state stores with the same name.
This is going to cause problems for some of the existing IQ APIs which require only a name to resolve the underlying state store. We're now going to need to take in the topology name in addition to the state store name to be able to locate the specific store a user wants to query
Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>
Introduces changelog headers to pass position information
to standby and restoring stores. The headers are guarded by an internal
config, which defaults to `false` for backward compatibility. Once IQv2
is finalized, we will make that flag a public config.
Reviewers: Patrick Stuedi <pstuedi@apache.org>, John Roesler <vvcephei@apache.org>
When the unit tests of the internal topic manager test
are executed on a slow machine (like sometimes in automatic builds)
they sometimes fail with a timeout exception instead of the expected
exception. To fix this behavior, this commit replaces the use of
system time with mock time.
Reviewer: John Roesler <vvcephei@apache.org>
Implements the major part of the IQv2 framework as proposed in KIP-796.
Reviewers: Patrick Stuedi <pstuedi@apache.org>, Vicky Papavasileiou <vpapavasileiou@confluent.io>, Bruno Cadonnna <cadonna@apache.org>
Log messages were changed in the AssignorConfiguration (#11490) that are
also used for verification in system test
StreamsCooperativeRebalanceUpgradeTest.test_upgrade_to_cooperative_rebalance.
This commit fixes the test and adds comments to the log messages
that point to the test that needs to be updated in case of
changes to the log messages.
Reviewers: John Roesler <vvcephei@apache.org>, Luke Chen <showuon@gmail.com>, David Jacot <djacot@confluent.io>
With a nice mock in RocksDBMetricsRecorderTest#shouldCorrectlyHandleHitRatioRecordingsWithZeroHitsAndMisses() and RocksDBMetricsRecorderTest#shouldCorrectlyHandleAvgRecordingsWithZeroSumAndCount() were green although getTickerCount() was never called. The tests were green because EasyMock returns 0 for a numerical return value by default if no expectation is specified. Thus, commenting out the expectation for getTickerCount() did not change the result of the test.
This commit changes the mock to a default mock and fixes the expectation to expect getAndResetTickerCount(). Now, commenting out the expectation leads to a test failure.
Reviewers: Luizfrf3 <lf.fonseca@hotmail.com>, Guozhang Wang <wangguoz@gmail.com>
Deprecate eager rebalancing protocol in kafka streams and log warning message when upgrade.from is set to 2.3 or lower. Also add a note in upgrade doc to prepare users for the removal of eager rebalancing support
Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>
Add additional asserts for KafkaStreamsTest: testInitializesAndDestroysMetricsReporters to help diagnose if it flakily fails in the future.
- MockMetricsReporter gets initialized only once during KafkaStreams construction, so make assert check stricter by ensuring initDiff is one.
- Assert KafkaStreams is not running before we validate whether MockMetricsMetricsReporter close count got incremented after streams close.
Reviewer: Bruno Cadonna <cadonna@apache.org>
Most configs that are read and used by Streams today originate from the properties passed in to the KafkaStreams constructor, which means they get applied universally across all threads, tasks, subtopologies, and so on. The only current exception to this is the topology.optimization config which is parsed from the properties that get passed in to StreamsBuilder#build. However there are a handful of configs that could also be scoped to the topology level, allowing users to configure each NamedTopology independently of the others, where it makes sense to do so.
This PR refactors the handling of these configs by interpreting the values passed in via KafkaStreams constructor as the global defaults, which can then be overridden for individual topologies via the properties passed in when building the NamedTopology. More topology-level configs may be added in the future, but this PR covers the following:
max.task.idle.ms
task.timeout.ms
buffered.records.per.partition
default.timestamp.extractor.class
default.deserialization.exception.handler
Reviewers: Guozhang Wang <wangguoz@gmail.com>, Walker Carlson <wcarlson@confluent.io>, Luke Chen <showuon@confluent.io>
* Migrate KTable joins to new Processor API.
* Migrate missing KTableProcessorSupplier implementations.
* Replace KTableProcessorSupplier with new Processor API implementation.
Reviewers: John Roesler <vvcephei@apache.org>
This PR adds some RocksDB metrics that could not be added in KIP-471 due to RocksDB version. The new metrics are extracted using Histogram data provided by RocksDB API, and the old ones were extracted using Tickers. The new metrics added are memtable-flush-time-(avg|min|max) and compaction-time-(avg|min|max).
Reviewer: Bruno Cadonnna <cadonna@apache.org>
Minor followup to #11405 / KIP-783 to write down the new guarantees we're providing about the meaning of a StreamsException in the javadocs of that class
Reviewers: Bruno Cadonna <cadonna@apache.org>
As raised in KAFKA-12994, All tests that use the old API should be either eliminated or migrated to the new API in order to remove the @SuppressWarnings("deprecation") annotations.
This PR migrates the SlidingWindowsTest to the new API.
Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>
To help users distinguish which task an exception was thrown from, and which NamedTopology if it exists, we add a TaskId field to the StreamsException class. We then make sure that all exceptions thrown to the handler are wrapped as StreamsExceptions, to help the user simplify their handling code as they know they will always need to unwrap the thrown exception exactly once.
Reviewers: Walker Carlson <wcarlson@confluent.io>, Luke Chen <showuon@gmail.com>, Guozhang Wang <wangguoz@gmail.com>, John Roesler <vvcephei@apache.org>
Disallow calling grace() if it was already set via ofTimeDifferenceAndGrace/WithNoGrace(). Add the check to disallow grace called after grace set via new API, and add tests for them.
Reviewers: Guozhang Wang <wangguoz@gmail.com>
We observed on the broker side that txn-offset-commit request with empty topics are received. After checking the source code I found there's on place on Streams which is unnecessarily sending empty offsets. This PR cleans up the streams layer logic a bit to not send empty offsets, and at the same time also guard against empty offsets at the producer layer as well.
Reviewers: Luke Chen <showuon@gmail.com>, Jason Gustafson <jason@confluent.io>
As part of the migration of KStream/KTable operations to the new
Processor API https://issues.apache.org/jira/browse/KAFKA-8410,
this PR includes the migration of KStream aggregate/reduce operations.
Reviewers: John Roesler <vvcephei@apache.org>
Recently a user hit this TaskAssignmentException due to a bug in their regex that meant no topics matched the pattern subscription, which in turn meant that it was impossible to resolve the number of partitions of the downstream repartition since there was no upstream topic to get the partition count for. Debugging this was pretty difficult and ultimately came down to stepping through the code line by line, since even with TRACE logging we only got a partial picture.
We should expand the logging to make sure the TRACE logging hits both conditional branches, and improve the error message with a suggestion for what to look for should someone hit this in the future
Reviewers: Walker Carlson <wcarlson@confluent.io>, Guozhang Wang <guozhang@confluent.io>, Matthias J. Sax <matthias@confluent.io>
In #9139, we added backward iterator on SessionStore. But there is a bug that when fetch/backwardFetch the key range, if there are multiple records in the same session window, we can't return the data in the correct order.
Reviewers: Anna Sophie Blee-Goldman <ableegoldman<apache.org>
This PR aims to utilize HighAvailabilityTaskAssignor to avoid downtime on corrupted tasks. The idea is that, when we hit TaskCorruptedException on an active task, a rebalance is triggered after we've wiped out the corrupted state stores. This will allow the assignor to temporarily redirect this task to another client who can resume work on the task while the original owner works on restoring the state from scratch.
Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>
As raised in KAFKA-12994, All tests that use the old API should be either eliminated or migrated to the new API in order to remove the @SuppressWarnings("deprecation") annotations. This PR will migrate over all the relevant tests in TimeWindowsTests.java
Reviewers: Anna Sophie Blee-Goldman
As part of the migration of KStream/KTable operations to the new Processor API https://issues.apache.org/jira/browse/KAFKA-8410, this PR includes the migration of KTable aggregate/reduce operations.
Reviewers: John Roesler <vvcephei@apache.org>
Add support for infinite range query for WindowStore. Story JIRA: https://issues.apache.org/jira/browse/KAFKA-13210
Reviewers: Patrick Stuedi <pstuedi@gmail.com>, Guozhang Wang <wangguoz@gmail.com>
KAFKA-13246: StoreQueryIntegrationTest#shouldQueryStoresAfterAddingAndRemovingStreamThread does not gate on stream state well
The test now waits for the client to transition to REBALANCING/RUNNING after adding/removing a thread as well as to transition to RUNNING before querying the state store.
Reviewers: singingMan <@3schwartz>, Walker Carlson <wcarlson@confluent.io>, John Roesler <vvcephei@apache.org>
This is to make sure that even if logging is disabled, we would still return null in order to workaround the deserialization issue for stream-stream left/outer joins.
Reviewers: Matthias J. Sax <mjsax@apache.org>
This is an alternative approach in parallel to #11235. After several unsuccessful trials to improve its efficiency i've come up with a larger approach, which is to use a kv-store instead as the shared store, which would store the value as list. The benefits of this approach are:
Only serde once that compose <timestamp, byte, key>, at the outer metered stores, with less byte array copies.
Deletes are straight-forward with no scan reads, just a single call to delete all duplicated <timestamp, byte, key> values.
Using a KV store has less space amplification than a segmented window store.
The cons though:
Each put call would be a get-then-write to append to the list; also we would spend a few more bytes to store the list (most likely a singleton list, and hence just 4 more bytes).
It's more complicated definitely.. :)
The main idea is that since the shared store is actively GC'ed by the expiration logic, not based on time retention, and since that the key format is in <timestamp, byte, key>, the range expiration query is quite efficient as well.
Added testing covering for the list stores (since we are still use kv-store interface, we cannot leverage on the get() calls in the stream-stream join, instead we use putIfAbsent and range only). Another minor factoring piggy-backed is to let toList to always close iterator to avoid leaking.
Reviewers: Sergio Peña <sergio@confluent.io>, Matthias J. Sax <mjsax@apache.org>
When introducing backward iterator for WindowStroe in #9138, we forgot to make "each segment" in reverse order (i.e. in descendingMap) in InMemoryWindowStore. Fix it and add integration tests for it.
Currently, in Window store, we store records in [segments -> [records] ].
For example:
window size = 500,
input records:
key: "a", value: "aa", timestamp: 0 ==> will be in [0, 500] window
key: "b", value: "bb", timestamp: 10 ==> will be in [0, 500] window
key: "c", value: "cc", timestamp: 510 ==> will be in [500, 1000] window
So, internally, the "a" and "b" will be in the same segment, and "c" in another segments.
segments: [0 /* window start */, records], [500, records].
And the records for window start 0 will be "a" and "b".
the records for window start 500 will be "c".
Before this change, we did have a reverse iterator for segments, but not in "records". So, when doing backwardFetchAll, we'll have the records returned in order: "c", "a", "b", which should be "c", "b", "a" obviously.
Reviewers: Jorge Esteban Quilcate Otoya <quilcate.jorge@gmail.com>, Anna Sophie Blee-Goldman <ableegoldman@apache.org>, Guozhang Wang <wangguoz@gmail.com>
When using EOS checkpointed offsets are not updated to the latest offsets from the changelog because the maybeWriteCheckpoint method is only ever called when commitNeeded=false. This change will force the update if enforceCheckpoint=true .
I have also added a test which verifies that both the state store and the checkpoint file are completely up to date with the changelog after the app has shutdown.
Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>, Guozhang Wang <wangguoz@gmail.com>
KAFKA-13243: KIP-773 Differentiate metric latency measured in ms and ns
Implementation of KIP-773
Deprecates inconsistent metrics bufferpool-wait-time-total,
io-waittime-total, and iotime-total.
Introduces new metrics bufferpool-wait-time-ns-total,
io-wait-time-ns-total, and io-time-ns-total with the same semantics as
before.
Adds metrics (old and new) in ops.html.
Adds upgrade guide for these metrics.
Reviewers: Guozhang Wang <wangguoz@gmail.com>, Tom Bentley <tbentley@redhat.com>
Description
Streams disables the write-ahead log (WAL) provided by RocksDB since it replicates the data in changelog topics. Hence, it does not make much sense to set WAL-related configs for RocksDB.
Proposed solution
Ignore any WAL-related configuration and state in the log that we are ignoring them.
Co-authored-by: Tomer Wizman <tomer.wizman@personetics.com>
Co-authored-by: Bruno Cadonna <cadonna@apache.org>
Reviewers: Boyang Chen <boyang@apache.org>, Bruno Cadonna <cadonna@apache.org>
As detailed in KAFKA-12994, unit tests using the old API should be either removed or migrated to the new API.
This PR migrates relevant tests in JoinWindowsTest.java and SessionWindowsTest.java.
Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>
Migrate Suppress as part of the migration of KStream/KTable
operations to the new Processor API (KAFKA-8410)
Reviewers: John Roesler <vvcephei@apache.org>
Currently KStreamMap and KStreamFlatMap classes will throw NPE if the call to KeyValueMapper#apply() return null. This commit checks whether the result of KeyValueMapper#apply() is null and throws a more meaningful error message for better debugging.
Two unit tests are also added to check if we successfully captured nulls.
Reviewers: Josep Prat <josep.prat@aiven.io>, Luke Chen <showuon@gmail.com>, Bruno Cadonna <cadonna@apache.org>
Instead of letting all RuntimeExceptions go through and be processed by the uncaught exception handler, IllegalStateException and IllegalArgumentException are not passed through and fail fast. In this PR when setting the uncaught exception handler we check if the exception is in an "exclude list", if so, we terminate the client, otherwise we continue as usual.
Added test checking this new case. Added integration test checking that user defined exception handler is not used when an IllegalStateException is thrown.
Reviewers: Bruno Cadonna <cadonna@confluent.io>, Guozhang Wang <wangguoz@gmail.com>
* Add the following producer metrics:
flush-time-total: cumulative sum of time elapsed during in flush.
txn-init-time-total: cumulative sum of time elapsed during in initTransactions.
txn-begin-time-total: cumulative sum of time elapsed during in beginTransaction.
txn-send-offsets-time-total: cumulative sum of time elapsed during in sendOffsetsToTransaction.
txn-commit-time-total: cumulative sum of time elapsed during in commitTransaction.
txn-abort-time-total: cumulative sum of time elapsed during in abortTransaction.
* Add the following consumer metrics:
commited-time-total: cumulative sum of time elapsed during in committed.
commit-sync-time-total: cumulative sum of time elapsed during in commitSync.
* Add a total-blocked-time metric to streams that is the sum of:
consumer’s io-waittime-total
consumer’s iotime-total
consumer’s committed-time-total
consumer’s commit-sync-time-total
restore consumer’s io-waittime-total
restore consumer’s iotime-total
admin client’s io-waittime-total
admin client’s iotime-total
producer’s bufferpool-wait-time-total
producer's flush-time-total
producer's txn-init-time-total
producer's txn-begin-time-total
producer's txn-send-offsets-time-total
producer's txn-commit-time-total
producer's txn-abort-time-total
Reviewers: Bruno Cadonna <cadonna@confluent.io>, Guozhang Wang <wangguoz@gmail.com>
Add a new case to the list of possible retriable exceptions for the flaky tests to take care of threads starting up
Reviewers: Leah Thomas <lthomas@confluent.io>, Anna Sophie Blee-Goldman
The GraphNode#writeToTopology method accepts a Properties input parameter, but never uses it in any of its implementations. We can remove this parameter to clean things up and help make it clear that writing nodes to the topology doesn't involve the app properties.
Reviewers: Bruno Cadonna <cadonna@confluent.io>
Currently in OrderedBytes#upperRange method, we'll check key bytes 1 by 1, to see if there's a byte value >= first timestamp byte value, so that we can skip the following key bytes, because we know compareTo will always return 0 or 1. However, in most cases, the first timestamp byte is always 0, more specifically the upperRange is called for both window store and session store. For former, the suffix is in timestamp, Long.MAX_VALUE and for latter the suffix is in Long.MAX_VALUE, timestamp. For Long.MAX_VALUE the first digit is not 0, for timestamp it could be 0 or not, but as long as it is up to "now" (i.e. Aug. 23rd) then the first byte should be 0 since the value is far smaller than what a long typed value could have. So in practice for window stores, that suffix's first byte has a large chance to be 0, and hence worth optimizing for.
This PR optimizes the not all query cases by not checking the key byte 1 by 1 (because we know the unsigned integer will always be >= 0), instead, put all bytes and timestamp directly. So, we won't have byte array copy in the end either.
Reviewers: Guozhang Wang <wangguoz@gmail.com>
The alive stream threads metric relies on the threads field as a monitor object for
its synchronized block. When the alive stream threads metric is registered it isn't
initialised so any call to get the metric value before it is initialised will result
in a null pointer exception.
Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>, Walker Carlson <wcarlson@confluent.io>
We increased the default session timeout to 30s in KIP-735:
https://cwiki.apache.org/confluence/display/KAFKA/KIP-735%3A+Increase+default+consumer+session+timeout
Since then, we are observing sporadic system test failures
due to rebalances taking longer than the test timeout.
Rather than increase the test wait times, we can just override
the session timeout to a value more appropriate in the testing
domain.
Reviewers: A. Sophie Blee-Goldman <ableegoldman@apache.org>
1. This test is taking two iterations since the firs iteration is designed to fail due to unknow topic leader. However both the timeout and the backoff are set to 50ms, while the actual SYSTEM time is used. This means we are likely to timeout before executing the second iteration. I thought about using a mock time but dropped that idea as it may forgo the purpose of this test, instead I set the backoff time to 10ms so that we are almost unlikely to hit this error anymore.
2. Found a minor issue while fixing this which is that when we have non-empty not-ready topics, but the topics-to-create is empty (which is possible as the test shouldCreateTopicWhenTopicLeaderNotAvailableAndThenTopicNotFound itself illustrates), we still call an empty create-topic function. Though it does not incur any round-trip it would still waste some cycles, so I branched it off and hence also simplified some unit tests.
Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>, Bruno Cadonna <cadonna@confluent.io>
Update to KIP-740.
Reviewers: A. Sophie Blee-Goldman <sophie@confluent.io>, Matthias J. Sax <matthias@confluent.io>, Konstantine Karantasis <konstantine@confluent.io>, Israel Ekpo <israelekpo@gmail.com>
We've seen a few failures recently due to the driver finishing
the production of data and verifying the results before the
whole cluster is even running.
Reviewers: Leah Thomas <lthomas@confluent.io>, Walker Carlson <wcarlson@confluent.io>, Matthias J. Sax <mjsax@apache.org>
Pt. 1: #10609
Pt. 2: #10683
Pt. 3: #10788
In Pt. 3 we implement the addNamedTopology API. This can be used to update the processing topology of a running Kafka Streams application without resetting the app, or even pausing/restarting the process. It's up to the user to ensure that this API is called on every instance of an application to ensure all clients are able to run the newly added NamedTopology.
Reviewers: Guozhang Wang <guozhang@confluent.io>, Walker Carlson <wcarlson@confluent.io>
Addresses the handful of remaining feedback from Pt. 2, plus adds two new tests: one verifying a multi-topology application with a FKJ and its internal topics, another to make sure IQ works with named topologies (though note that there is a bit more work left for IQ to be fully supported, will be tackled after Pt. 3
Reviewers: Guozhang Wang <guozhang@confluent.io>, Walker Carlson <wcarlson@confluent.io>
Add support to open endpoint range queries in key-value stores
Implements: KIP-763
Reviewers: Almog Gavra <almog@confluent.io>, Luke Chen <showuon@gmail.com>, John Roesler <vvcephei@apache.org>
Fix a bug in StoreQueryIntegrationTest::shouldQueryOnlyActivePartitionStoresByDefault that causes the test to fail in the case of a client rebalancing. The changes in this PR make sure the test keeps re-trying after a rebalancing operation, instead of failing.
Reviewers: Luke Chen <showuon@gmail.com>, John Roesler <vvcephei@apache.org>
As part of the migration of KStream/KTable operations to the new Processor API (KAFKA-8410), this PR includes the migration of KTable:
* mapValues,
* passthrough,
* and source operations.
Reviewers: John Roesler <vvcephei@apache.org>
I noticed that replace thread actions would not be logged unless the user added a log in the handler. I think it would be very useful for debugging.
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Pt. 1: #10609
Pt. 2: #10683
Pt. 3: #10788
The TopologyMetadata is next up after Pt. 1 #10609. This PR sets up the basic architecture for running an app with multiple NamedTopologies, though the APIs to add/remove them dynamically are not implemented until Pt. 3
Reviewers: Guozhang Wang <guozhang@confluent.io>, Walker Carlson <wcarlson@confluent.io>
Just a bit of minor cleanup that (a) does some prepwork for another PR I'm working on, (b) updates the javadocs & exception messages to report a more useful error to the user and describe what they actually need to do, and (c) hopefully makes these state checks more future-proof by defining methods for each kind of check in one place that can be easily updated instead of tracking down every individual check.
Reviewers: Walker Carlson <wcarlson@confluent.io>, Luke Chen <showuon@gmail.com>
Setting the max.poll.interval.ms to MAX_VALUE causes overflow when computing the joinGroupTimeoutMs and results in the JoinGroup timeout being set to the request.timeout.ms instead, which is much lower.
This can easily make consumers drop out of the group, since they must rejoin now within 30s (by default) yet have no obligation to almost ever call poll() given the high max.poll.interval.ms, especially when each record takes a long time to process or the `max.poll.records` is also very large. We just need to check for overflow and fix it to Integer.MAX_VALUE when it occurs.
Reviewers: Luke Chen <showuon@gmail.com>, John Roesler <vvcephei@apache.org>
There were a few followup things to address from #10926, most importantly a number of updates to the javadocs. Also includes a few missing verification checks.
Reviewers: Guozhang Wang <guozhang@confluent.io>, Matthias J. Sax <mjsax@apache.org>, Israel Ekpo
This test is flaky due to waiting on all records to be processed for only a single key before issuing IQ lookups and asserting whether data was found.
Reviewers: Phil Hardwick, Walker Carlson <wcarlson@confluent.io>
This test is hitting pretty frequent timeouts after bouncing a node and waiting for it to come back and fully rejoin the group. It seems to now take 45s for the initial JoinGroup to succeed, which I suspect is due to the new default session.interval.ms (which was recently changed to 45s). Let's try fixing this config to the old value of 10s and see if that helps it rejoin in time.
Reviewer: Bruno Cadonna <cadonna@apache.org>
When a new thread is added the queryable store providers continues to use the store providers it was given when KafkaStreams was instantiated. This means IQ will start performing lookups against an out-of-date list of threads, and may eventually become completely broken. We must make sure the QueryableStoreProvider is updated when threads are added and removed.
Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>
After changing the default serde to be null, some system tests started failing. This test didn't explicitly pass in a serde and didn't set the default config so when the test was trying to setup the source node it wasn't able to find any config to use and threw a config exception.
Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>, Bruno Cadonna <cadonna@confluent.io>
We removed default 24 hours grace period in KIP-633, and deprecate some grace methods, but we forgot to update the stream docs.
Reviewer: Bruno Cadonna <cadonna@apache.org>
Currently, we verify the startup of a Streams client by checking the transition
from REBALANCING to RUNNING and if the client processed some records
in the EOS system test. However, if the Streams client only
has standby tasks assigned as it can happen if the client is catching
up by using warm-up replicas, the client will never process
records within the timeout of the startup verification. Hence, the test
will fail although everything is fine. This commit fixes this by reducing
the time to the next probing rebalance and by increasing the number of
max warm-up replicas. In such a way, the catch up of the client and the
following processing of records should still be within the startup verification
timeout of the client.
Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>
If there is a cooperative rebalance the tasks might not be assigned to a thread at all for a very short timeframe, causing this test to fail. We can just retry getting the metadata until the group has finished rebalancing and all tasks are assigned
Reviewers: Bruno Cadonna <cadonna@apache.org>, Anna Sophie Blee-Goldman <ableegoldman@apache.org>, Josep Prat <josep.prat@aiven.io>
Follow up to #10917
The fix from #10917 intended to reduce the emit frequency to save the creation cost of RocksDB iterators. However, we incorrectly initialized the "timer" with timestamp zero, and thus, the timer was always in the past and we did try to emit left/outer join result too often.
This PR fixes the initialization of the emit interval timer to current wall-clock time to effectively 'enable' the fix from #10917.
Reviewers: Sergio Peña <sergio@confluent.io>, Guozhang Wang <guozhang@confluent.io>
In 2.8 and before, we computed the default grace period with Math.max(maintainDurationMs - sizeMs, 0); in method gracePeriodMs() in TimeWindows, SessionWindows, and JoinWindows. That means that the default grace period has never been 24 hours but 24 hours - window size. Since gracePeriodMs() is used to compute the retention time of the changelog topic for the corresponding window state store and the segments for the window state store it is important to keep the same computation for the deprecated methods. Otherwise, Streams app that run with 2.8 and before might not be compatible with Streams 3.0 because the retention time of the changelog topics created with older Streams apps will be smaller than the assumed retention time for Streams apps in 3.0. For example, with a window size of 10 hours, an old Streams app would have created a changelog topic with retention time 10 hours (window size) + 14 hours (default grace period, 24 hours - 10 hours). A 3.0 Streams app would assume a retention time of 10 hours (window size) + 24 hours (deprecated default grace period as currently specified on trunk). In the presence of failures, where a state store needs to recreated, records might get lost, because before the failure the state store of a 3.0 Streams app contained 10 hours + 24 hours of records whereas the changelog topic that was created with the old Streams app would only contain 10 hours + 14 hours of records.
All this happened due to us always stating that the default grace period was 24 hours although it was not completely correct and a connected and unfortunate misunderstanding when we removed deprecated windows APIs (#10378).
Co-authors: Bruno Cadonna <cadonna@apache.org>
Reviewers: Luke Chen <showuon@gmail.com>, Matthias J. Sax <mjsax@apache.org>, Bruno Cadonna <cadonna@apache.org>
Also update the upgrade guide indicating about the grace period KIP and its indication on the fix with throughput impact.
Reviewers: Luke Chen <showuon@gmail.com>, Matthias J. Sax <mjsax@apache.org>
The new prefixScan API may still throw UnsupportedVersionOperationException due to some missing implementations in vast store hierarchy of Streams, this PR adds those missing overrides and expands the test coverage.
Reviewers: Matthias J. Sax <matthias@confluent.io>, Anna Sophie Blee-Goldman <ableegoldman@apache.org>
Consolidate the RocksDBStoreTest and RocksDBKeyValueStoreTest files into a single test class for the RocksDBStore.
Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>
Update the docs for task idling, since the semantics have
changed in 3.0.
Reviewers: Jim Galasyn <jim.galasyn@confluent.io>, Luke Chen <showuon@gmail.com>, Boyang Chen <boyang@apache.org>
Log the specific StreamThreads participating in the rebalance for each client in the Streams application
Reviewers: Walker Carlson <wcarlson@confluent.io>, John Roesler <vvcephei@apache.org>
We changed the default serde in Streams to be null in #10813, but forgot to add some in tests, for example TestTopicsTest and TopologyTestDriverTest.
Reviewers: David Jacot <djacot@confluent.io>, Bruno Cadonna <cadonna@apache.org>
The fix to avoid spurious left/outer stream-stream join results, showed
very low throughput for RocksDB, due to excessive creation of iterators.
Instead of trying to emit left/outer stream-stream join result for every
input record, this PR adds tracking of the lower timestamp bound of
left/outer join candidates, and only tries to emit them (and create an
iterator) if they are potentially old enough.
Reviewers: Luke Chen <showuon@gmail.com>, Guozhang Wang <guozhang@confluent.io>, Sergio Peña <sergio@confluent.io>
This patch adds two new apis to support topic deletion using topic IDs or names. It uses a new class `TopicCollection` to keep a collection of topics defined either by names or IDs. Finally, it modifies `DeleteTopicsResult` to support both names and IDs and deprecates the old methods which have become ambiguous. Eventually we will want to deprecate the old `deleteTopics` apis as well, but this patch does not do so.
Reviewers: Jason Gustafson <jason@confluent.io>
Implements KIP-633.
Grace-period is an important parameter and its best to make it the user's responsibility to set it expliclity. Thus, we move off to provide a default and make it a mandatory parameter when creating a window.
Reviewers: A. Sophie Blee-Goldman <sophie@confluent.io>, Luke Chen <showuon@gmail.com>, Matthias J. Sax <matthias@confluent.io>
The integration test TaskMetadataIntegrationTest will increase
the length of the app ID when its test methods are called multiple
times in one execution. This is for example the case if you
repeatedly run the test until failure in IntelliJ IDEA. This might
also lead to exceptions because the state directory depends on the
app ID and directory names have a length limit.
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Session windows should not be close directly when "window end" time is reached, but "window close" time should be "window-end + gap + grace-period".
Reviewer: Matthias J. Sax <matthias@confluent.io>
Implementation of KIP-744.
Creates new Interfaces for TaskMetadata, ThreadMetadata, and
StreamsMetadata, providing internal implementations for each of them.
Deprecates current TaskMetadata, ThreadMetadata under o.a.k.s.processor,
and SreamsMetadata under a.o.k.s.state.
Updates references on internal classes from deprecated classes to new interfaces.
Deprecates methods on KafkaStreams returning deprecated ThreadMeatada and
StreamsMetadata, and provides new ones returning the new interfaces.
Update Javadocs referencing to deprecated classes and methods to point
to the right ones.
Co-authored-by: Bruno Cadonna <cadonna@apache.org>
Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>, Bruno Cadonna <cadonna@apache.org>
Throw an exception if a state directory exists as a regular file
Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>, Luke Chen <showuon@gmail.com>
* Deprecate the old Processor API
* Suppress warnings on all internal usages of the old API
(which will be migrated in other child tickets of KAFKA-8410)
* Add new KStream#process methods, since KAFKA-10603 has not seen any action.
This is a clean-up that we missed for "KIP-743: Remove config value 0.10.0-2.4 of Streams built-in metrics version config"
Reviewers: Luke Chen <showuon@gmail.com>, Matthias J. Sax <mjsax@apache.org>
They have been deprecated since 0.10.0. Full list of removes configs:
* port
* host.name
* advertised.port
* advertised.host.name
Also adjust tests to take the removals into account. Some tests were
no longer relevant and have been removed.
Finally, took the chance to:
* Clean up unnecessary usage of `KafkaConfig$.MODULE$` in
related files.
* Add missing `Test` annotations to `AdvertiseBrokerTest` and
make necessary changes for the tests to pass.
Reviewers: David Jacot <djacot@confluent.io>, Luke Chen <showuon@gmail.com>
We changed the behavior of left/outer stream-stream join via KAFKA-10847.
To avoid a breaking change during an upgrade, we need to disable this
fix by default.
We only enable the fix if users opt-in expliclity by changing their
code. We leverage KIP-633 (KAFKA-8613) that offers a new JoinWindows
API with mandatory grace-period to enable the fix.
Reviewers: Sergio Peña <sergio@confluent.io>, Israel Ekpo <israelekpo@gmail.com>, Guozhang Wang <guozhang@confluent.io>
Many Kafka protocol JSON files were accidentally configured to not use
flexible versions, since it was not on by default. This PR requires
JSON files to specify a flexibleVersions value. If the JSON file does
not specify the flexibleVersions value, display an error message
suggesting the correct value to use for new messages.
Reviewers: Jason Gustafson <jason@confluent.io>
This IT has been failing on trunk recently. Enabling EOS during the integration test
makes it easier to be sure that the test's assumptions are really true during verification
and should make the test more reliable.
I also noticed that in the actual system test file, we are using the deprecated property
name "beta" instead of "v2".
Reviewers: Boyang Chen <boyang@apache.org>
While there might still be some issue about the test as described here by @ableegoldman , but I found the reason why this test failed quite frequently recently. It's because we increased the session timeout to 45 sec in KIP-735.
The reason why increasing session timeout affected this test is because in this test, we will keep adding new stream clients and remove old one, to maintain only 3 stream clients alive. The problem here is, when old stream closed, we won't trigger rebalance immediately due to the stream clients are all static members as described in KIP-345, which means, we will trigger trigger group rebalance only when session.timeout expired. That said, when old client closed, we'll have at least 45 sec with some tasks not working.
Also, in this test, we have 2 timeout conditions to fail this test before verification passed:
1. 6 minutes timeout
2. polling 30 times (each with 5 seconds) without getting any data. (that is, 5 * 30 = 150 sec without consuming any data)
For (1), in my test under 45 session timeout, we'll create 8 stream clients, which means, we'll have 5 clients got closed. And each closed client need 45 sec to trigger rebalance, so we'll have 45 * 5 = 225 sec (~4 mins) of the time having some tasks not working.
For (2), during new client created and old client closed, it need some time to do rebalance. With 45 session timeout, we only got ~100 sec left. In slow jenkins env, it might reach the 30 retries without getting any data timeout.
Therefore, decreasing session timeout can make this test completes faster and more reliable.
Reviewers: Guozhang Wang <wangguoz@gmail.com>
1. Update StateDirectory#clean
- Delete application's statestore directory in cleanup process if it is empty.
2. Add Tests
- StateDirectoryTest#shouldDeleteAppDirWhenCleanUpIfEmpty: asserting the empty application directory is deleted with StateDirectory#clean.
- StateDirectoryTest#shouldNotDeleteAppDirWhenCleanUpIfNotEmpty: asserting the non-empty application directory is not deleted with StateDirectory#clean and appropriate log message is generated.
- Add Integration test: StateDirectoryIntegrationTest
3. Improve EOSUncleanShutdownIntegrationTest: test all available cases regarding cleanup process on unclean shutdown.
Reviewers: John Roesler <vvcephei@apache.org>, Guozhang Wang <guozhang@apache.org>
Remove unused methods in internal classes
Mark fields that can be final as final
Remove unneeded generic type annotation
Convert single use fields to local final variables
Use method reference in lambdas when it's more readable
Reviewers: Matthias J. Sax <mjsax@apache.org>, Bruno Cadonna <cadonna@apache.org>
* Development of EasyMock and PowerMock has stagnated while Mockito continues to be actively developed. With the new Java cadence, it's a problem to depend on libraries that do bytecode generation and are not actively maintained. In addition, Mockito is also easier to use.KAFKA-7438
Reviewers: Ismael Juma <ismael@juma.me.uk>, Chia-Ping Tsai <chia7712@gmail.com>, Bruno Cadonna <cadonna@apache.org>
This PR includes adding the NamedTopology to the Subscription/AssignmentInfo, and to the StateDirectory so it can place NamedTopology tasks within the hierarchical structure with task directories under the NamedTopology parent dir.
Reviewers: Walker Carlson <wcarlson@confluent.io>, Guozhang Wang <guozhang@confluent.io>
1. When register state stores, add the store to globalStateStores before calling any blocking calls that may throw errors, so that upon closing we would close the stores as well.
2. Remove the other list as a local field, and call topology.globalStateStores when needed to get the list.
Reviewers: A. Sophie Blee-Goldman <sophie@confluent.io>, Bruno Cadonna <cadonna@apache.org>
Refactored logConfig to be passed appropriately when using shutDownWhenFull or emitEarlyWhenFull. Removed the constructor that doesn't accept a logConfig parameter so you're forced to specify it explicitly, whether it's empty/unspecified or not.
Co-authored-by: Bruno Cadonna <cadonna@apache.org>
Reviewers: Walker Carlson <wcarlson@confluent.io>, Bruno Cadonna <cadonna@apache.org>
As specified in KIP-743, this PR removes the built-in metrics
in Streams that are superseded by the refactoring proposed in KIP-444.
Reviewers: Guozhang Wang <wangguoz@gmail.com>, Luke Chen <showuon@gmail.com>
* Lay the groundwork for migrating KTable Processors to the new PAPI.
* Migrate the KTableFilter processor to prove that the groundwork works.
This is an effort to help break up #10507 into multiple PRs.
Reviewers: Boyang Chen <boyang@apache.org>
Quick followup to KIP-740 to actually deprecate this constructor, and update the upgrade guide with what we changed in KIP-740. I also noticed the TaskId#parse method had been modified previously, and should be re-added to the public TaskId class. It had no tests, so now it does
Reviewers: Matthias J. Sax <mjsax@confluent.io>, Luke Chen <showuon@gmail.com>
This PR tries to add the check for transaction timeout for a comparison against commit interval of streams. If transaction timeout is smaller than commit interval, stream should crash and inform user to update their commit interval to be larger or equal to the given transaction timeout, or vise versa.
Reviewers: Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <wangguoz@gmail.com>, A. Sophie Blee-Goldman <sophie@confluent.io>
As described in KIP-740, we clean up the public TaskId class and introduce new APIs to return it from TaskMetadata
Reviewers: Guozhang Wang <guozhang@confluent.io>
Increase session timeout to fix flaky KTableKTableForeignKeyInnerJoinMultiIntegrationTest.shouldInnerJoinMultiPartitionQueryable
Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>
In #10676 we renamed the internal Subtopology class that implemented the TopologyDescription.Subtopology interface. By mistake, we also renamed the interface itself, which is a public API. This wasn't really the intended point of that PR, so rather than do a retroactive KIP, let's just reverse the renaming.
Reviewers: Walker Carlson <wcarlson@confluent.io>, Guozhang Wang <guozhang@confluent.io>
Minor followup to #10573. Removes this internal Producer config which was only ever used to avoid a very minor amount of work to downgrade the consumer group metadata in the txn commit request in Kafka Streams
Reviewers: Ismael Juma <ismael@juma.me.uk>, Matthias J. Sax <mjsax@confluent.io>, Guozhang Wang <guozhang@confluent.io>
This PR aims to streamline the configurations for WindowedDeserialisers as described in KIP-725. It deprecates default.windowed.key.serde.inner and default.windowed.value.serde.inner configs in StreamConfig and adds windowed.inner.class.serde.
Reviewers: Anna Sophie Blee-Goldman<ableegoldman@apache.org>
Currently KafkaStreams#cleanUp only throw an IllegalStateException if the state is RUNNING or REBALANCING, however the application could be in the process of shutting down in which case StreamThreads may still be running. We should also throw if the state is PENDING_ERROR or PENDING_SHUTDOWN
Reviewers: Walker Carlson <wcarlson@confluent.io>, Guozhang Wang <guozhang@confluent.io>
Introduce List serde for primitive types or custom serdes with a serializer and a deserializer according to KIP-466
Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>, Matthias J. Sax <mjsax@conflunet.io>, John Roesler <roesler@confluent.io>, Michael Noll <michael@confluent.io>
Consecutive UUID generation could result in same prefix.
Reviewers: Josep Prat <josep.prat@aiven.io>, Anna Sophie Blee-Goldman <ableegoldman@apache.org>
This PR upgrades RocksDB to 6.19.3. After the upgrade the Gradle build exited with code 134 due to SIGABRT signals ("Pure virtual function called!") coming from the C++ part of RocksDB. This error was caused by RocksDB state stores not properly closed in Streams' code. This PR adds the missing closings and updates the RocksDB option adapter.
Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>, Guozhang Wang <wangguoz@gmail.com>
We currently use hamcrest imports to check the outputs of the RelationalSmokeTest, but with the new gradle updates, the proper hamcrest imports are no longer included in the test jar.
This is a bit of a workaround to remove the hamcrest usage so we can get system tests up and running again. Potential follow-up could be to update the way we create the test-jar to pull in the proper dependencies.
Reviewers: Bruno Cadonna <cadonna@apache.org>
Try to address the extreme flakiness of shouldInnerJoinMultiPartitionQueryable since the recent test cleanup. Since we need to wait for 3 streams reach RUNNING state, it makes sense to increase the waiting time to make the test more reliable.
Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>
When users supply in-memory stores for left/outer joins, then the internal shared outer store must be switch to in-memory store too. This will allow users who want to keep all stores in memory to continue doing so.
Added unit tests to validate topology and left/outer joins work fine with an in-memory shared store.
Reviewers: Guozhang Wang <wangguoz@gmail.com>
The test cases for ThreaCache didn't have the corresponding unit tests for all, reverseAll and reverseRange methods. This PR aims to add the same.
Reviewers: Bruno Cadonna <cadonna@apache.org>
Adds an internal flag that can be used to disable the fixes in KAFKA-10847. It defaults to true if the flag is not set or has an invalid boolean value.
The flag is named __enable.kstreams.outer.join.spurious.results.fix__. This flag is considered internal only. It is a temporary flag that will be used to help users to disable the join fixes while they do a transition from the previous semantics of left/outer joins. The flag may be removed in future releases.
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Minor followup to #10342 that I noticed while working on the NamedTopology stuff. Cleans up a few things:
We no longer need locking for the global state directory either, since it's contained within the top-level state directory lock. Definitely less critical than the task directory locking, since it's less vulnerable to IOExceptions given that it's just locked and unlocked once during the application lifetime, but nice to have nonetheless
Clears out misc. usages of the LOCK_FILE_NAME that no longer apply. This has the awesome side effect of finally being able to actually delete obsolete task directories, whereas previously we had to leave behind the empty directory due to a ridiculous Windows bug (though I'm sure they would claim "it's not a bug it's a feature" 😉 )
Lazily delete old-and-now-unused lock files in the StateDirectory#taskDirIsEmpty method to clean up the state directory for applications that upgraded from an older version that still used task locking
Reviewers: Walker Carlson <wcarlson@confluent.io>
Fixes the issue with https://issues.apache.org/jira/browse/KAFKA-10847.
To fix the above problem, the left/outer stream-stream join processor uses a buffer to hold non-joined records for some time until the window closes, so they are not processed if a join is found during the join window time. If the window of a record closes and a join was not found, then this should be emitted and processed by the consequent topology processor.
A new time-ordered window store is used to temporary hold records that do not have a join and keep the records keys ordered by time. The KStreamStreamJoin has a reference to this new store . For every non-joined record seen, the processor writes it to this new state store without processing it. When a joined record is seen, the processor deletes the joined record from the new state store to prevent further processing.
Records that were never joined at the end of the window + grace period are emitted to the next topology processor. I use the stream time to check for the expiry time for determinism results . The KStreamStreamJoin checks for expired records and emit them every time a new record is processed in the join processor.
The new state store is shared with the left and right join nodes. The new store needs to serialize the record keys using a combined key of <joinSide-recordKey>. This key combination helps to delete the records from the other join if a joined record is found. Two new serdes are created for this, KeyAndJoinSideSerde which serializes a boolean value that specifies the side where the key is found, and ValueOrOtherValueSerde that serializes either V1 or V2 based on where the key was found.
Reviewers: Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <wangguoz@gmail.com>