[jira] [Created] (KAFKA-16699) Have Streams treat InvalidPidMappingException like a ProducerFencedException
Walker Carlson created KAFKA-16699: -- Summary: Have Streams treat InvalidPidMappingException like a ProducerFencedException Key: KAFKA-16699 URL: https://issues.apache.org/jira/browse/KAFKA-16699 Project: Kafka Issue Type: Improvement Components: streams Reporter: Walker Carlson -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Created] (KAFKA-16316) Make the restore behavior of GlobalKTables with custom processors configureable
Walker Carlson created KAFKA-16316: -- Summary: Make the restore behavior of GlobalKTables with custom processors configureable Key: KAFKA-16316 URL: https://issues.apache.org/jira/browse/KAFKA-16316 Project: Kafka Issue Type: Improvement Components: streams Reporter: Walker Carlson Assignee: Walker Carlson Take the change implemented in https://issues.apache.org/jira/browse/KAFKA-7663 and make it optional through adding a couple methods to the API -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Resolved] (KAFKA-14936) Add Grace Period To Stream Table Join
[ https://issues.apache.org/jira/browse/KAFKA-14936?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Walker Carlson resolved KAFKA-14936. Resolution: Done > Add Grace Period To Stream Table Join > - > > Key: KAFKA-14936 > URL: https://issues.apache.org/jira/browse/KAFKA-14936 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: Walker Carlson >Assignee: Walker Carlson >Priority: Major > Labels: kip, streams > Fix For: 3.6.0 > > > Include the grace period for stream table joins as described in kip 923. > Also add a rocksDB time based queueing implementation of > `TimeOrderedKeyValueBuffer` -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Created] (KAFKA-15379) Add option for Grace period Joins to disable changelog creation
Walker Carlson created KAFKA-15379: -- Summary: Add option for Grace period Joins to disable changelog creation Key: KAFKA-15379 URL: https://issues.apache.org/jira/browse/KAFKA-15379 Project: Kafka Issue Type: New Feature Reporter: Walker Carlson Right now if you are preforming a buffered join with a grace period there is no way to avoid the creation of a changelog -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Created] (KAFKA-14936) Add Grace Period To Stream Table Join
Walker Carlson created KAFKA-14936: -- Summary: Add Grace Period To Stream Table Join Key: KAFKA-14936 URL: https://issues.apache.org/jira/browse/KAFKA-14936 Project: Kafka Issue Type: Improvement Components: streams Reporter: Walker Carlson Assignee: Walker Carlson Include the grace period for stream table joins as described in kip 923. Also add a rocksDB time based queueing implementation of `TimeOrderedKeyValueBuffer` -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Created] (KAFKA-13676) When processing in ALOS we might as well commit progress made other tasks on a task specific exception
Walker Carlson created KAFKA-13676: -- Summary: When processing in ALOS we might as well commit progress made other tasks on a task specific exception Key: KAFKA-13676 URL: https://issues.apache.org/jira/browse/KAFKA-13676 Project: Kafka Issue Type: Improvement Components: streams Reporter: Walker Carlson Assignee: Walker Carlson When processing in ALOS we might as well commit progress made other tasks on a task specific exception. If one task has an issue and we have already successfully completed processing on at least one task it would be good to commit those successfully processed tasks. This should prevent limit the duplicated records downstream and also be more efficient. Also if one task is having lots of issues the other tasks can at least make progress. When we introduced the thread replacement mechanism this optimization became possible. -- This message was sent by Atlassian Jira (v8.20.1#820001)
[jira] [Created] (KAFKA-13588) We should consolidate `changelogFor` methods to simplify the generation of internal topic names
Walker Carlson created KAFKA-13588: -- Summary: We should consolidate `changelogFor` methods to simplify the generation of internal topic names Key: KAFKA-13588 URL: https://issues.apache.org/jira/browse/KAFKA-13588 Project: Kafka Issue Type: Improvement Components: streams Reporter: Walker Carlson [https://github.com/apache/kafka/pull/11611#discussion_r772625486] we should use `ProcessorContextUtils#changelogFor` after we remove `init(final ProcessorContext context, final StateStore root)` in `CahceingWindowStore#initInternal` Or any other place that we generate an internal topic name. -- This message was sent by Atlassian Jira (v8.20.1#820001)
[jira] [Created] (KAFKA-13246) StoreQueryIntegrationTest#shouldQueryStoresAfterAddingAndRemovingStreamThread does not gate on stream state well
Walker Carlson created KAFKA-13246: -- Summary: StoreQueryIntegrationTest#shouldQueryStoresAfterAddingAndRemovingStreamThread does not gate on stream state well Key: KAFKA-13246 URL: https://issues.apache.org/jira/browse/KAFKA-13246 Project: Kafka Issue Type: Improvement Components: streams Reporter: Walker Carlson StoreQueryIntegrationTest#shouldQueryStoresAfterAddingAndRemovingStreamThread should be improved by waiting for the client to go to rebalancing or running after adding and removing a thread. It should also wait until running before querying the state store -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Resolved] (KAFKA-13215) Flaky test org.apache.kafka.streams.integration.TaskMetadataIntegrationTest.shouldReportCorrectEndOffsetInformation
[
https://issues.apache.org/jira/browse/KAFKA-13215?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Walker Carlson resolved KAFKA-13215.
Resolution: Fixed
> Flaky test
> org.apache.kafka.streams.integration.TaskMetadataIntegrationTest.shouldReportCorrectEndOffsetInformation
> ---
>
> Key: KAFKA-13215
> URL: https://issues.apache.org/jira/browse/KAFKA-13215
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Reporter: Konstantine Karantasis
>Assignee: Walker Carlson
>Priority: Major
> Labels: flaky-test
> Fix For: 3.1.0
>
>
> Integration test {{test
> org.apache.kafka.streams.integration.TaskMetadataIntegrationTest.shouldReportCorrectCommittedOffsetInformation()}}
> sometimes fails with
> {code:java}
> java.lang.AssertionError: only one task
> at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:26)
> at
> org.apache.kafka.streams.integration.TaskMetadataIntegrationTest.getTaskMetadata(TaskMetadataIntegrationTest.java:163)
> at
> org.apache.kafka.streams.integration.TaskMetadataIntegrationTest.shouldReportCorrectEndOffsetInformation(TaskMetadataIntegrationTest.java:144)
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Created] (KAFKA-12781) Improve the endOffsets accuracy in TaskMetadata
Walker Carlson created KAFKA-12781: -- Summary: Improve the endOffsets accuracy in TaskMetadata Key: KAFKA-12781 URL: https://issues.apache.org/jira/browse/KAFKA-12781 Project: Kafka Issue Type: Improvement Components: streams Reporter: Walker Carlson Currently `TaskMetadata#endOffsets()` returns the highest offset seen by the consumer so far. It should be possible to get the highest offset in the topic via the consumer instead. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (KAFKA-12754) TaskMetadata endOffsets does not update when the offsets are read
Walker Carlson created KAFKA-12754: -- Summary: TaskMetadata endOffsets does not update when the offsets are read Key: KAFKA-12754 URL: https://issues.apache.org/jira/browse/KAFKA-12754 Project: Kafka Issue Type: Bug Components: streams Reporter: Walker Carlson Assignee: Walker Carlson The high water mark in StreamTask is not updated optimally. Also it would be good to have the metadata offsets have a initial value of -1 instead of an empty map that way the set of TopicPartitions won't change. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (KAFKA-12711) Add a back off option to Replace thread
Walker Carlson created KAFKA-12711: -- Summary: Add a back off option to Replace thread Key: KAFKA-12711 URL: https://issues.apache.org/jira/browse/KAFKA-12711 Project: Kafka Issue Type: Improvement Components: streams Reporter: Walker Carlson There should be a native option to set a back off when replacing a thread. Either there should be a config and a user chosen strategy or a value you can set in the handler that causes a delay in creating the new thread. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (KAFKA-12705) Task idling is not sufficiently tested
Walker Carlson created KAFKA-12705: -- Summary: Task idling is not sufficiently tested Key: KAFKA-12705 URL: https://issues.apache.org/jira/browse/KAFKA-12705 Project: Kafka Issue Type: Improvement Components: streams Reporter: Walker Carlson The test for task idling are a bit sparse. When I changed it so that isProcessable always returns true only one test failed. That means the entire code path is hinging on one unit test (shouldBeProcessableIfAllPartitionsBuffered) that does not cover all branches of logic. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (KAFKA-12699) Streams no longer overrides the java default uncaught exception handler
Walker Carlson created KAFKA-12699: -- Summary: Streams no longer overrides the java default uncaught exception handler Key: KAFKA-12699 URL: https://issues.apache.org/jira/browse/KAFKA-12699 Project: Kafka Issue Type: Bug Components: streams Affects Versions: 2.8.0 Reporter: Walker Carlson If a user used `Thread.setUncaughtExceptionHanlder()` to set the handler for all threads in the runtime streams would override that with its own handler. However since streams does not use the `Thread` handler anymore it will no longer do so. This can cause problems if the user does something like `System.exit(1)` in the handler. If using the old handler in streams it will still work as it used to -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (KAFKA-12691) TaskMetadata timeSinceIdlingStarted not reporting correctly
Walker Carlson created KAFKA-12691: -- Summary: TaskMetadata timeSinceIdlingStarted not reporting correctly Key: KAFKA-12691 URL: https://issues.apache.org/jira/browse/KAFKA-12691 Project: Kafka Issue Type: Bug Components: streams Reporter: Walker Carlson Assignee: Walker Carlson TaskMetadata timeSinceIdlingStarted not reporting correctly. It takes into account suspended but not the call to is processable. To fix this we need to record when the first time it is not processable. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (KAFKA-12565) Global thread only topologies should be able to shutdown applications via the uncaught exception handler
Walker Carlson created KAFKA-12565: -- Summary: Global thread only topologies should be able to shutdown applications via the uncaught exception handler Key: KAFKA-12565 URL: https://issues.apache.org/jira/browse/KAFKA-12565 Project: Kafka Issue Type: Improvement Components: streams Affects Versions: 3.0.0, 2.8.0 Reporter: Walker Carlson Global thread only topologies should be able to shutdown applications via the uncaught exception handler. Currently because there is no stream thread in this case there is nothing to participate in a rebalance to communicate the request. If we add a stream thread to do this it will result in an `IllegalStateException` because "Consumer is not subscribed to any topics or assigned any partitions". -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (KAFKA-12538) Global Threads should be able to be replaced like stream threads
Walker Carlson created KAFKA-12538: -- Summary: Global Threads should be able to be replaced like stream threads Key: KAFKA-12538 URL: https://issues.apache.org/jira/browse/KAFKA-12538 Project: Kafka Issue Type: Improvement Components: streams Reporter: Walker Carlson We should be able to replace global threads from the streams uncaught exception handler just like we replace stream threads. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (KAFKA-12537) Single Threaded EOS applications will not work with SHUTDOWN_APPLICATION
Walker Carlson created KAFKA-12537: -- Summary: Single Threaded EOS applications will not work with SHUTDOWN_APPLICATION Key: KAFKA-12537 URL: https://issues.apache.org/jira/browse/KAFKA-12537 Project: Kafka Issue Type: Bug Affects Versions: 3.0.0, 2.8.0 Reporter: Walker Carlson Single Threaded EOS applications will not work with the streams uncaught exception handler option SHUTDOWN_APPLICATION. This is because the EOS thread needs to close and clean up, but to send the shutdown signal it needs to have at least one thread running. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (KAFKA-12503) Resizing the thread cache in a non thread safe way can cause records to be redirected throughout the topology
Walker Carlson created KAFKA-12503: -- Summary: Resizing the thread cache in a non thread safe way can cause records to be redirected throughout the topology Key: KAFKA-12503 URL: https://issues.apache.org/jira/browse/KAFKA-12503 Project: Kafka Issue Type: Bug Components: streams Affects Versions: 2.8.0 Reporter: Walker Carlson Fix For: 2.8.0 When a thread is added, removed or replaced the cache is resized. When the thread cache was resized it was being done so from the thread initiating these calls. This can cause the record to be redirected to the wrong processor via the call to `evict` in the cache. The evict flushes records downstream to the next processor after the cache. But if this is on the wrong thread the wrong processor receives them. This can cause 3 problems. 1) When the owner finishes processing the record it set the current node to null in the processor context a this then causes the other processor to throw an exception `StreamsException: Current node is unknown.`. 2) Depending on the type it can cause a class cast exception as the record is a different type. Mostly this happened when the value types were different inside of the map node from the toStream method 3) A silent issue is it could cause data to be processed by the wrong node and cause data corruption. We have not been able to confirm this last one but it is the most dangerous in many ways. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (KAFKA-12462) Threads in PENDING_SHUTDOWN entering a rebalance can cause an illegal state exception
Walker Carlson created KAFKA-12462:
--
Summary: Threads in PENDING_SHUTDOWN entering a rebalance can
cause an illegal state exception
Key: KAFKA-12462
URL: https://issues.apache.org/jira/browse/KAFKA-12462
Project: Kafka
Issue Type: Bug
Reporter: Walker Carlson
A thread was removed, sending it to the PENDING_SHUTDOWN state, but went
through a rebalance before completing the shutdown.
{code:java}
// [2021-03-07 04:33:39,385] DEBUG [i-07430efc31ad166b7-StreamThread-6]
stream-thread [i-07430efc31ad166b7-StreamThread-6] Ignoring request to transit
from PENDING_SHUTDOWN to PARTITIONS_REVOKED: only DEAD state is a valid next
state (org.apache.kafka.streams.processor.internals.StreamThread)
{code}
Inside StreamsRebalanceListener#onPartitionsRevoked, we have
{code:java}
//
if (streamThread.setState(State.PARTITIONS_REVOKED) != null &&
!partitions.isEmpty())
taskManager.handleRevocation(partitions);
{code}
Since PENDING_SHUTDOWN → PARTITIONS_REVOKED is a disallowed transition, we
never invoke TaskManager#handleRevocation. Currently handleRevocation is
responsible for preparing any active tasks for close, including committing
offsets and writing the checkpoint as well as suspending the task. We can’t
close the task in handleRevocation since we still support EAGER rebalancing,
which invokes handleRevocation at the beginning of a rebalance on all tasks.
The tasks that are actually revoked will be closed during
TaskManager#handleAssignment . The IllegalStateException is specifically
because we don’t suspend the task before attempting to close it, and the direct
transition from RUNNING → CLOSED is forbidden.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Reopened] (KAFKA-12347) Improve Kafka Streams ability to track progress
[ https://issues.apache.org/jira/browse/KAFKA-12347?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Walker Carlson reopened KAFKA-12347: > Improve Kafka Streams ability to track progress > --- > > Key: KAFKA-12347 > URL: https://issues.apache.org/jira/browse/KAFKA-12347 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: Walker Carlson >Assignee: Walker Carlson >Priority: Major > Labels: kip > Fix For: 3.0.0 > > > Add methods to track records being consumed fully and to tell if tasks are > idling. This will allow users of streams to build uptime metrics around > streams with less difficulty. > KIP-715: > [https://cwiki.apache.org/confluence/display/KAFKA/KIP-715%3A+Expose+Committed+offset+in+streams] -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Resolved] (KAFKA-12362) Determine if a Task is idling
[ https://issues.apache.org/jira/browse/KAFKA-12362?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Walker Carlson resolved KAFKA-12362. Resolution: Abandoned > Determine if a Task is idling > - > > Key: KAFKA-12362 > URL: https://issues.apache.org/jira/browse/KAFKA-12362 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: Walker Carlson >Priority: Major > Fix For: 3.0.0 > > > determine if a task is idling given the task Id. > > https://github.com/apache/kafka/pull/10180 -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (KAFKA-12362) Determine if a Task is idling
Walker Carlson created KAFKA-12362: -- Summary: Determine if a Task is idling Key: KAFKA-12362 URL: https://issues.apache.org/jira/browse/KAFKA-12362 Project: Kafka Issue Type: Improvement Components: streams Reporter: Walker Carlson Fix For: 3.0.0 determine if a task is idling given the task Id. https://github.com/apache/kafka/pull/10180 -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (KAFKA-12347) Improve Kafka Streams ability to track progress
Walker Carlson created KAFKA-12347: -- Summary: Improve Kafka Streams ability to track progress Key: KAFKA-12347 URL: https://issues.apache.org/jira/browse/KAFKA-12347 Project: Kafka Issue Type: Improvement Components: streams Reporter: Walker Carlson Assignee: Walker Carlson Fix For: 3.0.0 -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Resolved] (KAFKA-4640) Improve Streams unit test coverage
[
https://issues.apache.org/jira/browse/KAFKA-4640?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Walker Carlson resolved KAFKA-4640.
---
Resolution: Fixed
> Improve Streams unit test coverage
> --
>
> Key: KAFKA-4640
> URL: https://issues.apache.org/jira/browse/KAFKA-4640
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 0.11.0.0
>Reporter: Damian Guy
>Assignee: Damian Guy
>Priority: Minor
> Attachments: streams-coverage.zip
>
>
> There are some important methods in streams that are lacking good unit-test
> coverage. Whilst we shouldn't strive to get 100% coverage, we should do our
> best to ensure sure that all important code paths are covered by unit-tests.
> For contributors: you can first run {{./gradlew streams:reportCoverage}} to
> get the report, which will then accessible in
> {{streams/build/reports/jacoco/test/html/index.html}}.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Resolved] (KAFKA-10015) React Smartly to Unexpected Errors on Stream Threads
[ https://issues.apache.org/jira/browse/KAFKA-10015?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Walker Carlson resolved KAFKA-10015. Resolution: Fixed > React Smartly to Unexpected Errors on Stream Threads > > > Key: KAFKA-10015 > URL: https://issues.apache.org/jira/browse/KAFKA-10015 > Project: Kafka > Issue Type: Improvement > Components: streams >Affects Versions: 2.5.0 >Reporter: Bruno Cadonna >Assignee: Walker Carlson >Priority: Major > Labels: kip > > Currently, if an unexpected error occurs on a stream thread, the stream > thread dies, a rebalance is triggered, and the Streams' client continues to > run with less stream threads. > Some errors trigger a cascading of stream thread death, i.e., after the > rebalance that resulted from the death of the first thread the next thread > dies, then a rebalance is triggered, the next thread dies, and so forth until > all stream threads are dead and the instance shuts down. Such a chain of > rebalances could be avoided if an error could be recognized as the cause of > cascading stream deaths and as a consequence the Streams' client could be > shut down after the first stream thread death. > On the other hand, some unexpected errors are transient and the stream thread > could safely be restarted without causing further errors and without the need > to restart the Streams' client. > The goal of this ticket is to classify errors and to automatically react to > the errors in a way to avoid cascading deaths and to recover stream threads > if possible. > KIP-663: > [https://cwiki.apache.org/confluence/display/KAFKA/KIP-663%3A+API+to+Start+and+Shut+Down+Stream+Threads] -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (KAFKA-12247) Make removeStreamThread work better with static membership
Walker Carlson created KAFKA-12247: -- Summary: Make removeStreamThread work better with static membership Key: KAFKA-12247 URL: https://issues.apache.org/jira/browse/KAFKA-12247 Project: Kafka Issue Type: Improvement Reporter: Walker Carlson Assignee: Walker Carlson Fix For: 2.8.0 Ensure that calling removeStreamThread make the thread leave the group right away instead of waiting for the timeout. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (KAFKA-12184) Once the Error Classification is better update the default streams uncaught exception handler to replace threads
Walker Carlson created KAFKA-12184: -- Summary: Once the Error Classification is better update the default streams uncaught exception handler to replace threads Key: KAFKA-12184 URL: https://issues.apache.org/jira/browse/KAFKA-12184 Project: Kafka Issue Type: Improvement Components: streams Reporter: Walker Carlson -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (KAFKA-10810) Add a replace thread option to the streams uncaught exception handler
Walker Carlson created KAFKA-10810: -- Summary: Add a replace thread option to the streams uncaught exception handler Key: KAFKA-10810 URL: https://issues.apache.org/jira/browse/KAFKA-10810 Project: Kafka Issue Type: Improvement Reporter: Walker Carlson Assignee: Walker Carlson Add an option to replace threads that have died. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (KAFKA-10705) Avoid World Readable RocksDB
Walker Carlson created KAFKA-10705: -- Summary: Avoid World Readable RocksDB Key: KAFKA-10705 URL: https://issues.apache.org/jira/browse/KAFKA-10705 Project: Kafka Issue Type: Bug Reporter: Walker Carlson The state directory could be protected more restrictive by preventing access to state directory for group and others. At least other should have no readable access -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (KAFKA-9299) Over eager optimization
Walker Carlson created KAFKA-9299:
-
Summary: Over eager optimization
Key: KAFKA-9299
URL: https://issues.apache.org/jira/browse/KAFKA-9299
Project: Kafka
Issue Type: Task
Components: streams
Reporter: Walker Carlson
There are a few cases where the optimizer will attempt an optimization that can
cause a copartitioning failure. Known case of this are related to join and
cogroup, however could also effect merge or others.
Take for example three input topics A, B and C with 2, 3 and 4 partitions
respectively.
B' = B.map();
B'.join(A)
B'.join(C)
the optimizer will push up the repartition upstream and with will cause the
copartitioning to fail.
Can be seen with the following test:
@Test
public void shouldInsertRepartitionsTopicForCogroupsUsedTwice() {
final StreamsBuilder builder = new StreamsBuilder();
final Properties properties = new Properties();
properties.setProperty(StreamsConfig.TOPOLOGY_OPTIMIZATION,
StreamsConfig.OPTIMIZE);
final KStream stream1 = builder.stream("one",
stringConsumed);
final KGroupedStream groupedOne = stream1.map((k, v) ->
new KeyValue<>(v, k)).groupByKey(Grouped.as("foo"));
final CogroupedKStream one =
groupedOne.cogroup(STRING_AGGREGATOR);
one.aggregate(STRING_INITIALIZER);
one.aggregate(STRING_INITIALIZER);
final String topologyDescription =
builder.build(properties).describe().toString();
System.err.println(topologyDescription);
}
Topologies:
Sub-topology: 0
Source: KSTREAM-SOURCE-00 (topics: [one])
--> KSTREAM-MAP-01
Processor: KSTREAM-MAP-01 (stores: [])
--> foo-repartition-filter
<-- KSTREAM-SOURCE-00
Processor: foo-repartition-filter (stores: [])
--> foo-repartition-sink
<-- KSTREAM-MAP-01
Sink: foo-repartition-sink (topic: foo-repartition)
<-- foo-repartition-filter
Sub-topology: 1
Source: foo-repartition-source (topics: [foo-repartition])
--> COGROUPKSTREAM-AGGREGATE-06,
COGROUPKSTREAM-AGGREGATE-12
Processor: COGROUPKSTREAM-AGGREGATE-06 (stores:
[COGROUPKSTREAM-AGGREGATE-STATE-STORE-02])
--> COGROUPKSTREAM-MERGE-07
<-- foo-repartition-source
Processor: COGROUPKSTREAM-AGGREGATE-12 (stores:
[COGROUPKSTREAM-AGGREGATE-STATE-STORE-08])
--> COGROUPKSTREAM-MERGE-13
<-- foo-repartition-source
Processor: COGROUPKSTREAM-MERGE-07 (stores: [])
--> none
<-- COGROUPKSTREAM-AGGREGATE-06
Processor: COGROUPKSTREAM-MERGE-13 (stores: [])
--> none
<-- COGROUPKSTREAM-AGGREGATE-12
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Created] (KAFKA-9298) Reuse of a mapped stream causes an Invalid Topology
Walker Carlson created KAFKA-9298:
-
Summary: Reuse of a mapped stream causes an Invalid Topology
Key: KAFKA-9298
URL: https://issues.apache.org/jira/browse/KAFKA-9298
Project: Kafka
Issue Type: Bug
Components: streams
Reporter: Walker Carlson
Can be found with in the KStreamKStreamJoinTest.java
@Test
public void optimizerIsEager() {
final StreamsBuilder builder = new StreamsBuilder();
final KStream stream1 = builder.stream("topic",
Consumed.with(Serdes.String(), Serdes.String()));
final KStream stream2 = builder.stream("topic2",
Consumed.with(Serdes.String(), Serdes.String()));
final KStream stream3 = builder.stream("topic3",
Consumed.with(Serdes.String(), Serdes.String()));
final KStream newStream = stream1.map((k, v) -> new
KeyValue<>(v, k));
newStream.join(stream2,
(value1, value2) -> value1 + value2,
JoinWindows.of(ofMillis(100)),
StreamJoined.with(Serdes.String(), Serdes.String(),
Serdes.String()));
newStream.join(stream3,
(value1, value2) -> value1 + value2,
JoinWindows.of(ofMillis(100)),
StreamJoined.with(Serdes.String(), Serdes.String(),
Serdes.String()));
System.err.println(builder.build().describe().toString());
}
results in
Invalid topology: Topic KSTREAM-MAP-03-repartition has already been
registered by another source.
org.apache.kafka.streams.errors.TopologyException: Invalid topology: Topic
KSTREAM-MAP-03-repartition has already been registered by another
source.
at
org.apache.kafka.streams.processor.internals.InternalTopologyBuilder.validateTopicNotAlreadyRegistered(InternalTopologyBuilder.java:578)
at
org.apache.kafka.streams.processor.internals.InternalTopologyBuilder.addSource(InternalTopologyBuilder.java:378)
at
org.apache.kafka.streams.kstream.internals.graph.OptimizableRepartitionNode.writeToTopology(OptimizableRepartitionNode.java:100)
at
org.apache.kafka.streams.kstream.internals.InternalStreamsBuilder.buildAndOptimizeTopology(InternalStreamsBuilder.java:303)
at
org.apache.kafka.streams.StreamsBuilder.build(StreamsBuilder.java:562)
at
org.apache.kafka.streams.StreamsBuilder.build(StreamsBuilder.java:551)
at
org.apache.kafka.streams.kstream.internals.KStreamKStreamJoinTest.optimizerIsEager(KStreamKStreamJoinTest.java:136)
at
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at
java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:566)
at
org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59)
at
org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
at
org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56)
at
org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:305)
at
org.junit.runners.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100)
at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:365)
at
org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:103)
at
org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:63)
at org.junit.runners.ParentRunner$4.run(ParentRunner.java:330)
at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:78)
at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:328)
at org.junit.runners.ParentRunner.access$100(ParentRunner.java:65)
at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:292)
at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:305)
at org.junit.runners.ParentRunner.run(ParentRunner.java:412)
at
org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.runTestClass(JUnitTestClassExecutor.java:110)
at
org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.execute(JUnitTestClassExecutor.java:58)
at
org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.execute(JUnitTestClassExecutor.java:38)
at
org.gradle.api.internal.tasks.testing.junit.AbstractJUnitTestClassProcessor.processTestClass(AbstractJUnitTestClassProcessor.java:62)
at
org.gradle.api.internal.tasks.testing.SuiteTestClassProcessor.processTestClass(SuiteTestClassProcessor.java:51)
at
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native
[jira] [Created] (KAFKA-9243) Update the javadocs from KeyValueStore to TimestampKeyValueStore
Walker Carlson created KAFKA-9243: - Summary: Update the javadocs from KeyValueStore to TimestampKeyValueStore Key: KAFKA-9243 URL: https://issues.apache.org/jira/browse/KAFKA-9243 Project: Kafka Issue Type: Improvement Components: admin, clients, consumer, KafkaConnect, producer , streams, streams-test-utils Reporter: Walker Carlson Materialized objects use KayValueStore but the docs should be for TimestampedKeyValueStore because of changes to Materialized. This tickets should be broken down in a series of smaller PRs to keep the scope of each PR contained, allowing for more effective reviews. -- This message was sent by Atlassian Jira (v8.3.4#803005)
