[jira] [Commented] (KAFKA-9533) ValueTransform forwards `null` values
[ https://issues.apache.org/jira/browse/KAFKA-9533?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034298#comment-17034298 ] Bruno Cadonna commented on KAFKA-9533: -- [~mviamari] Thank you for opening this ticket. This seems to be a bug indeed. > ValueTransform forwards `null` values > - > > Key: KAFKA-9533 > URL: https://issues.apache.org/jira/browse/KAFKA-9533 > Project: Kafka > Issue Type: Bug > Components: streams >Affects Versions: 2.4.0 >Reporter: Michael Viamari >Priority: Minor > > According to the documentation for `KStream#transformValues`, nulls returned > from `ValueTransformer#transform` are not forwarded. (see > [KStream#transformValues|https://kafka.apache.org/24/javadoc/org/apache/kafka/streams/kstream/KStream.html#transformValues-org.apache.kafka.streams.kstream.ValueTransformerSupplier-java.lang.String...-] > However, this does not appear to be the case. In > `KStreamTransformValuesProcessor#process` the result of the transform is > forwarded directly. > {code:java} > @Override > public void process(final K key, final V value) { > context.forward(key, valueTransformer.transform(key, value)); > } > {code} -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-4928) Add integration test for DumpLogSegments
[ https://issues.apache.org/jira/browse/KAFKA-4928?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034358#comment-17034358 ] ASF GitHub Bot commented on KAFKA-4928: --- Sammers21 commented on pull request #4145: KAFKA-4928: Add integration test for DumpLogSegments URL: https://github.com/apache/kafka/pull/4145 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Add integration test for DumpLogSegments > > > Key: KAFKA-4928 > URL: https://issues.apache.org/jira/browse/KAFKA-4928 > Project: Kafka > Issue Type: Test > Components: log, tools >Reporter: Ismael Juma >Priority: Major > Labels: newbie > > DumpLogSegments is an important tool to analyse log files, but we have no > JUnit tests for it. It would be good to have some tests that verify that the > output is sane for a populated log. > Our system tests call DumpLogSegments, but we should be able to detect > regressions via the JUnit test suite. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (KAFKA-9483) Add Scala KStream#toTable to the Streams DSL
[ https://issues.apache.org/jira/browse/KAFKA-9483?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] highluck updated KAFKA-9483: Description: [KIP-523: Add KStream#toTable to the Streams DSL|https://cwiki.apache.org/confluence/display/KAFKA/KIP-523%3A+Add+KStream%23toTable+to+the+Streams+DSL?src=jira] I am trying to add the same function to scala was: [KIP-523: Add KStream#toTable to the Streams DSL|https://cwiki.apache.org/confluence/display/KAFKA/KIP-523%3A+Add+KStream%23toTable+to+the+Streams+DSL?src=jira] I am trying to add the same function to scalar > Add Scala KStream#toTable to the Streams DSL > > > Key: KAFKA-9483 > URL: https://issues.apache.org/jira/browse/KAFKA-9483 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: highluck >Assignee: highluck >Priority: Minor > > [KIP-523: Add KStream#toTable to the Streams > DSL|https://cwiki.apache.org/confluence/display/KAFKA/KIP-523%3A+Add+KStream%23toTable+to+the+Streams+DSL?src=jira] > > I am trying to add the same function to scala -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-9455) Consider using TreeMap for in-memory stores of Streams
[ https://issues.apache.org/jira/browse/KAFKA-9455?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034478#comment-17034478 ] highluck commented on KAFKA-9455: - [~guozhang] I have a question. Are you referring to the following form of point queries? "WindowStoreIterator fetch(final Bytes key)" thank you! > Consider using TreeMap for in-memory stores of Streams > -- > > Key: KAFKA-9455 > URL: https://issues.apache.org/jira/browse/KAFKA-9455 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: Guozhang Wang >Assignee: highluck >Priority: Major > Labels: newbie++ > > From [~ableegoldman]: It's worth noting that it might be a good idea to > switch to TreeMap for different reasons. Right now the ConcurrentSkipListMap > allows us to safely perform range queries without copying over the entire > keyset, but the performance on point queries seems to scale noticeably worse > with the number of unique keys. Point queries are used by aggregations while > range queries are used by windowed joins, but of course both are available > within the PAPI and for interactive queries so it's hard to say which we > should prefer. Maybe rather than make that tradeoff we should have one > version for efficient range queries (a "JoinWindowStore") and one for > efficient point queries ("AggWindowStore") - or something. I know we've had > similar thoughts for a different RocksDB store layout for Joins (although I > can't find that ticket anywhere..), it seems like the in-memory stores could > benefit from a special "Join" version as well cc/ Guozhang Wang > Here are some random thoughts: > 1. For kafka streams processing logic (i.e. without IQ), it's better to make > all processing logic relying on point queries rather than range queries. > Right now the only processor that use range queries are, as mentioned above, > windowed stream-stream joins. I think we should consider using a different > window implementation for this (and as a result also get rid of the > retainDuplicate flags) to refactor the windowed stream-stream join operation. > 2. With 1), range queries would only be exposed as IQ. Depending on its usage > frequency I think it makes lots of sense to optimize for single-point queries. > Of course, even without step 1) we should still consider using tree-map for > windowed in-memory stores to have a better scaling effect. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-9515) Upgrade ZooKeeper to 3.5.7
[ https://issues.apache.org/jira/browse/KAFKA-9515?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034514#comment-17034514 ] Ron Dagostino commented on KAFKA-9515: -- [~ijuma] I have learned something since [~rsivaram] and I discussed this in the above-referenced PR. Specifically, if someone tries to run the ZK Security Migrator tool with no SASL creds but with a client cert, while at the same time ZooKeeper 3.5.7 specifies ssl.clientAuth=none, then the migrator tool fails and ACls are not applied. This is the stack trace I see: org.apache.zookeeper.KeeperException$InvalidACLException: KeeperErrorCode = InvalidACL for /admin at org.apache.zookeeper.KeeperException.create(KeeperException.java:128) at org.apache.zookeeper.KeeperException.create(KeeperException.java:54) at kafka.zookeeper.AsyncResponse.maybeThrow(ZooKeeperClient.scala:564) at kafka.zk.KafkaZkClient.createRecursive(KafkaZkClient.scala:1644) at kafka.zk.KafkaZkClient.makeSurePersistentPathExists(KafkaZkClient.scala:1566) at kafka.admin.ZkSecurityMigrator.$anonfun$run$4(ZkSecurityMigrator.scala:270) at kafka.admin.ZkSecurityMigrator.$anonfun$run$4$adapted(ZkSecurityMigrator.scala:265) at scala.collection.immutable.List.foreach(List.scala:392) at kafka.admin.ZkSecurityMigrator.kafka$admin$ZkSecurityMigrator$$run(ZkSecurityMigrator.scala:265) at kafka.admin.ZkSecurityMigrator$.run(ZkSecurityMigrator.scala:110) at kafka.admin.ZkSecurityMigrator$.main(ZkSecurityMigrator.scala:115) at kafka.admin.ZkSecurityMigrator.main(ZkSecurityMigrator.scala) So we don't have to worry about the case of someone thinking they applied ACLs when in fact they have not -- they can never be confused about that due to the above error being raised, so if ssl.clientAuth=none on the ZooKeeper side they must have set SASL credentials for it to succeed, and the SASL user will be ACL'ed in ZooKeeper. The other case to think about is after ACLs have been applied successfully and Kafka connects with zookeeper.set.acl=true. The possibilities here are they set a SASL credential with or without a client cert or they don't set a SASL credential with a client cert. If they set a SASL credential then all is good -- the SASL user has ben ACl'ed and it works fine. If they don't set a SASL credential then they are connecting with the client cert, and of course that won't be ACL'ed (and if ssl.clientAuth=none in ZooKeeper then they are connecting unauthenticated anyway) -- and Kafka will fail to start. I think that is fine because the thing we had to worry about is a potential breach of security: someone thinking they were setting up ZooKeeper with secure ACls but they actually ended up creating ACls for World read/write access. As we can see from the above, this is actually impossible. So assuming I haven't missed anything or made a mistake, I think the only thing we would need to do is some doc changes to let people know about ssl.clientAuth=[want|need|none] being operational. > Upgrade ZooKeeper to 3.5.7 > -- > > Key: KAFKA-9515 > URL: https://issues.apache.org/jira/browse/KAFKA-9515 > Project: Kafka > Issue Type: Improvement >Reporter: Ismael Juma >Assignee: Ismael Juma >Priority: Blocker > Fix For: 2.5.0, 2.4.1 > > > There are some critical fixes in ZK 3.5.7 and the first RC has been posted: > [https://mail-archives.apache.org/mod_mbox/zookeeper-dev/202002.mbox/%3cCAGH6_KiULzemT-V4x_2ybWeKLMvQ+eh=q-dzsiz8a-ypp5t...@mail.gmail.com%3e] -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-9515) Upgrade ZooKeeper to 3.5.7
[ https://issues.apache.org/jira/browse/KAFKA-9515?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034526#comment-17034526 ] Ismael Juma commented on KAFKA-9515: Can we get the doc changes ready please? > Upgrade ZooKeeper to 3.5.7 > -- > > Key: KAFKA-9515 > URL: https://issues.apache.org/jira/browse/KAFKA-9515 > Project: Kafka > Issue Type: Improvement >Reporter: Ismael Juma >Assignee: Ismael Juma >Priority: Blocker > Fix For: 2.5.0, 2.4.1 > > > There are some critical fixes in ZK 3.5.7 and the first RC has been posted: > [https://mail-archives.apache.org/mod_mbox/zookeeper-dev/202002.mbox/%3cCAGH6_KiULzemT-V4x_2ybWeKLMvQ+eh=q-dzsiz8a-ypp5t...@mail.gmail.com%3e] -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-9280) Duplicate messages are observed in ACK mode ALL
[ https://issues.apache.org/jira/browse/KAFKA-9280?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034540#comment-17034540 ] Stanislav Kozlovski commented on KAFKA-9280: [~vikram484] this shouldn't happen because the leader waits for a second fetch request that proves that the follower has that offset. e.g high watermark is 1000. leader and follower are both at 1000. The follower dies but managed to send a fetch request - FetchRequest\{from=1000} in flight. Meanwhile the producer produces offset 1001 with acks=all. The leader will not acknowledge that produce request until all in-sync followers issue a FetchRequest with a `from` value of at least 1001. Does that make sense? > Duplicate messages are observed in ACK mode ALL > --- > > Key: KAFKA-9280 > URL: https://issues.apache.org/jira/browse/KAFKA-9280 > Project: Kafka > Issue Type: Bug >Affects Versions: 2.2.1 >Reporter: VIkram >Priority: Major > > In ack mode ALL, leader is sending the message to consumer even before > receiving the acknowledgements from other replicas. This can lead to > *+duplicate messages+*. > > Setup details: > * 1 zookeeper, 5 brokers > * Producer: Synchronous > * Topic: 1 partition, replication factor - 3, min isr - 2 > > Say First replica (Leader), Second replica and Third replica are the three > replicas of the topic. > > *Sequence of events:* > a) All brokers are up and running. > b) Clients started running. > c) Kill second replica of the topic. > d) Kill the third replica. Now min isr will not be satisfied. > e) Bring up third replica. Min isr will be satisfied. > > *Breakdown of step 'd':* > # Just before producer sends next message, killed third replica with kill -9 > (Leader takes time ~5sec to detect that the broker is down). > # Producer sent a message to leader. > # Before the leader knows that third replica is down, it accepts the message > from producer. > # Leader forwards the message to third replica. > # Before receiving ACK from third replica, leader sent the message to > consumer. > # Leader doesn't get an ACK from third replica. > # Now leader detects that third replica is down and throws > NOT_ENOUGH_REPLICAS_EXCEPTION. > # Now leader stops accepting messages from producer. > # Since producer didn't get ACK from leader, it will throw TIMEOUT_EXCEPTION > after timeout (2min in our case) . > # So far, producer believes that the message was not received by leader > whereas the consumer actually received it. > # Now producer retries sending the same message. (In our application it is > the next integer we send). > # Now when second/third replica is up, leader accepts the message and sends > the same message to consumer. *Thus sending duplicates.* > > > *Logs:* > # 2-3 seconds before producer sends next message, killed third replica with > kill -9 (Leader takes time ~5sec to detect that the broker is down). > _{{{_ > _> kill -9 49596_ > _}}}_ > __ > # Producer sent a message to leader. > _{{{_ > _[20190327 11:02:48.231 EDT (main-Terminating-1) Will send message: > ProducerRecord(topic=t229, partition=null, headers=RecordHeaders(headers = > [], isReadOnly = false), key=null, value=p229-4, timestamp=null)_ > _}}}_ > > # Before the leader knows that third replica is down, it accepts the message > from producer. > # Leader forwards the message to third replica. > # Before receiving ACK from third replica, leader sent the message to > consumer. > _{{{_ > _Received MessageRecord: ConsumerRecord(topic = t229, partition = 0, > leaderEpoch = 1, offset = 3, CreateTime = 1553698968232, serialized key size > = -1, serialized value size = 6, headers = RecordHeaders(headers = [], > isReadOnly = false), key = null, value = p229-4)_ > _}}}_ > __ > # Leader doesn't get an ACK from third replica. > # Now leader detects that third replica is down and throws > NOT_ENOUGH_REPLICAS_EXCEPTION. > _{{{_ > _[2019-03-27 11:02:53,541] ERROR [ReplicaManager broker=0] Error processing > append operation on partition t229-0 (kafka.server.ReplicaManager)_ > _org.apache.kafka.common.errors.NotEnoughReplicasException: The size of the > current ISR Set(0) is insufficient to satisfy the min.isr requirement of 2 > for partition t229-0_ > _}}}_ > > # Now leader stops accepting messages from producer. > # Since producer didn't get ACK from leader, it will throw TIMEOUT_EXCEPTION > after timeout (2min in our case) . > _{{{_ > _java.util.concurrent.ExecutionException: > org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for > t229-0:12 ms_ > _has passed since batch creation_ > _at > org.apache.kafka.clients.producer.internals.FutureRecordMetadata.valueOrError(FutureRecordMetadata.java:98)_ > _at > org.apache.kafka.clie
[jira] [Commented] (KAFKA-9107) Change input Type of configBackingStore for StandaloneHerder
[ https://issues.apache.org/jira/browse/KAFKA-9107?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034570#comment-17034570 ] ASF GitHub Bot commented on KAFKA-9107: --- rhauch commented on pull request #7602: KAFKA-9107; Change constructor of StandaloneHerder of test version URL: https://github.com/apache/kafka/pull/7602 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Change input Type of configBackingStore for StandaloneHerder > > > Key: KAFKA-9107 > URL: https://issues.apache.org/jira/browse/KAFKA-9107 > Project: Kafka > Issue Type: Improvement > Components: KafkaConnect >Affects Versions: 2.3.0 >Reporter: Xu JianHai >Priority: Trivial > Fix For: 2.3.1 > > > In method descriptor of StandaloneHerder constructor (Test version), > configBackingStore should be ConfigBackingStore instead of > MemoryConfigBackingStore. > I have change and submit pr. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-7061) Enhanced log compaction
[ https://issues.apache.org/jira/browse/KAFKA-7061?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034592#comment-17034592 ] Jun Rao commented on KAFKA-7061: [~senthilm-ms] : Sorry for the delay. I will review the PR this week. > Enhanced log compaction > --- > > Key: KAFKA-7061 > URL: https://issues.apache.org/jira/browse/KAFKA-7061 > Project: Kafka > Issue Type: Improvement > Components: core >Affects Versions: 2.5.0 >Reporter: Luis Cabral >Assignee: Senthilnathan Muthusamy >Priority: Major > Labels: kip > > Enhance log compaction to support more than just offset comparison, so the > insertion order isn't dictating which records to keep. > Default behavior is kept as it was, with the enhanced approached having to be > purposely activated. > The enhanced compaction is done either via the record timestamp, by settings > the new configuration as "timestamp" or via the record headers by setting > this configuration to anything other than the default "offset" or the > reserved "timestamp". > See > [KIP-280|https://cwiki.apache.org/confluence/display/KAFKA/KIP-280%3A+Enhanced+log+compaction] > for more details. > +From Guozhang:+ We should emphasize on the WIKI that the newly introduced > config yields to the existing "log.cleanup.policy", i.e. if the latter's > value is `delete` not `compact`, then the previous config would be ignored. > +From Jun Rao:+ With the timestamp/header strategy, the behavior of the > application may need to change. In particular, the application can't just > blindly take the record with a larger offset and assuming that it's the value > to keep. It needs to check the timestamp or the header now. So, it would be > useful to at least document this. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-6793) Unnecessary warning log message
[ https://issues.apache.org/jira/browse/KAFKA-6793?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034616#comment-17034616 ] John Roesler commented on KAFKA-6793: - It seems like we could fix this simply by not passing invalid configurations to the clients, for example, we pass them prefixed configs, but we could just strip the prefixes off. > Unnecessary warning log message > > > Key: KAFKA-6793 > URL: https://issues.apache.org/jira/browse/KAFKA-6793 > Project: Kafka > Issue Type: Bug > Components: streams >Affects Versions: 1.1.0 >Reporter: Anna O >Priority: Minor > > When upgraded KafkaStreams from 0.11.0.2 to 1.1.0 the following warning log > started to appear: > level: WARN > logger: org.apache.kafka.clients.consumer.ConsumerConfig > message: The configuration 'admin.retries' was supplied but isn't a known > config. > The config is not explicitly supplied to the streams. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-9013) Flaky Test MirrorConnectorsIntegrationTest#testReplication
[ https://issues.apache.org/jira/browse/KAFKA-9013?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034661#comment-17034661 ] Matthias J. Sax commented on KAFKA-9013: [https://builds.apache.org/job/kafka-pr-jdk11-scala2.13/4631/testReport/junit/org.apache.kafka.connect.mirror/MirrorConnectorsIntegrationTest/testReplication/] > Flaky Test MirrorConnectorsIntegrationTest#testReplication > -- > > Key: KAFKA-9013 > URL: https://issues.apache.org/jira/browse/KAFKA-9013 > Project: Kafka > Issue Type: Bug > Components: KafkaConnect >Reporter: Bruno Cadonna >Priority: Major > Labels: flaky-test > > h1. Stacktrace: > {code:java} > java.lang.AssertionError: Condition not met within timeout 2. Offsets not > translated downstream to primary cluster. > at org.apache.kafka.test.TestUtils.waitForCondition(TestUtils.java:377) > at org.apache.kafka.test.TestUtils.waitForCondition(TestUtils.java:354) > at > org.apache.kafka.connect.mirror.MirrorConnectorsIntegrationTest.testReplication(MirrorConnectorsIntegrationTest.java:239) > {code} > h1. Standard Error > {code} > Standard Error > Oct 09, 2019 11:32:00 PM org.glassfish.jersey.internal.inject.Providers > checkProviderRuntime > WARNING: A provider > org.apache.kafka.connect.runtime.rest.resources.ConnectorPluginsResource > registered in SERVER runtime does not implement any provider interfaces > applicable in the SERVER runtime. Due to constraint configuration problems > the provider > org.apache.kafka.connect.runtime.rest.resources.ConnectorPluginsResource will > be ignored. > Oct 09, 2019 11:32:00 PM org.glassfish.jersey.internal.inject.Providers > checkProviderRuntime > WARNING: A provider > org.apache.kafka.connect.runtime.rest.resources.LoggingResource registered in > SERVER runtime does not implement any provider interfaces applicable in the > SERVER runtime. Due to constraint configuration problems the provider > org.apache.kafka.connect.runtime.rest.resources.LoggingResource will be > ignored. > Oct 09, 2019 11:32:00 PM org.glassfish.jersey.internal.inject.Providers > checkProviderRuntime > WARNING: A provider > org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource registered > in SERVER runtime does not implement any provider interfaces applicable in > the SERVER runtime. Due to constraint configuration problems the provider > org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource will be > ignored. > Oct 09, 2019 11:32:00 PM org.glassfish.jersey.internal.inject.Providers > checkProviderRuntime > WARNING: A provider > org.apache.kafka.connect.runtime.rest.resources.RootResource registered in > SERVER runtime does not implement any provider interfaces applicable in the > SERVER runtime. Due to constraint configuration problems the provider > org.apache.kafka.connect.runtime.rest.resources.RootResource will be ignored. > Oct 09, 2019 11:32:01 PM org.glassfish.jersey.internal.Errors logErrors > WARNING: The following warnings have been detected: WARNING: The > (sub)resource method listLoggers in > org.apache.kafka.connect.runtime.rest.resources.LoggingResource contains > empty path annotation. > WARNING: The (sub)resource method listConnectors in > org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource contains > empty path annotation. > WARNING: The (sub)resource method createConnector in > org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource contains > empty path annotation. > WARNING: The (sub)resource method listConnectorPlugins in > org.apache.kafka.connect.runtime.rest.resources.ConnectorPluginsResource > contains empty path annotation. > WARNING: The (sub)resource method serverInfo in > org.apache.kafka.connect.runtime.rest.resources.RootResource contains empty > path annotation. > Oct 09, 2019 11:32:02 PM org.glassfish.jersey.internal.inject.Providers > checkProviderRuntime > WARNING: A provider > org.apache.kafka.connect.runtime.rest.resources.ConnectorPluginsResource > registered in SERVER runtime does not implement any provider interfaces > applicable in the SERVER runtime. Due to constraint configuration problems > the provider > org.apache.kafka.connect.runtime.rest.resources.ConnectorPluginsResource will > be ignored. > Oct 09, 2019 11:32:02 PM org.glassfish.jersey.internal.inject.Providers > checkProviderRuntime > WARNING: A provider > org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource registered > in SERVER runtime does not implement any provider interfaces applicable in > the SERVER runtime. Due to constraint configuration problems the provider > org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource will be > ignored. > Oct 09, 2019 11:32:02 PM org.gla
[jira] [Created] (KAFKA-9537) Abstract transformations in configurations cause unfriendly error message.
Jeremy Custenborder created KAFKA-9537: -- Summary: Abstract transformations in configurations cause unfriendly error message. Key: KAFKA-9537 URL: https://issues.apache.org/jira/browse/KAFKA-9537 Project: Kafka Issue Type: Bug Components: KafkaConnect Affects Versions: 2.4.0 Reporter: Jeremy Custenborder Assignee: Jeremy Custenborder I was working with a coworker who had a bash script posting a config to connect with {code:java}org.apache.kafka.connect.transforms.ExtractField.$Key{code} in the script. Bash removed the $Key because it wasn't escaped properly. {code:java} org.apache.kafka.connect.transforms.ExtractField.{code} is made it to the rest interface. A Class was create for the abstract implementation of ExtractField and passed to getConfigDefFromTransformation. It tried to call newInstance which threw an exception. The following gets returned via the rest interface. {code} { "error_code": 400, "message": "Connector configuration is invalid and contains the following 1 error(s):\nInvalid value class org.apache.kafka.connect.transforms.ExtractField for configuration transforms.extractString.type: Error getting config definition from Transformation: null\nYou can also find the above list of errors at the endpoint `/{connectorType}/config/validate`" } {code} It would be a much better user experience if we returned something like {code} { "error_code": 400, "message": "Connector configuration is invalid and contains the following 1 error(s):\nInvalid value class org.apache.kafka.connect.transforms.ExtractField for configuration transforms.extractString.type: Error getting config definition from Transformation: Transformation is abstract and cannot be created.\nYou can also find the above list of errors at the endpoint `/{connectorType}/config/validate`" } {code} or {code} { "error_code": 400, "message": "Connector configuration is invalid and contains the following 1 error(s):\nInvalid value class org.apache.kafka.connect.transforms.ExtractField for configuration transforms.extractString.type: Error getting config definition from Transformation: Transformation is abstract and cannot be created. Did you mean ExtractField$Key, ExtractField$Value?\nYou can also find the above list of errors at the endpoint `/{connectorType}/config/validate`" } {code} -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-9181) Flaky test kafka.api.SaslGssapiSslEndToEndAuthorizationTest.testNoConsumeWithoutDescribeAclViaSubscribe
[ https://issues.apache.org/jira/browse/KAFKA-9181?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034677#comment-17034677 ] ASF GitHub Bot commented on KAFKA-9181: --- hachikuji commented on pull request #8084: KAFKA-9181; Maintain clean separation between local and group subscriptions in consumer's SubscriptionState (#7941) URL: https://github.com/apache/kafka/pull/8084 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Flaky test > kafka.api.SaslGssapiSslEndToEndAuthorizationTest.testNoConsumeWithoutDescribeAclViaSubscribe > --- > > Key: KAFKA-9181 > URL: https://issues.apache.org/jira/browse/KAFKA-9181 > Project: Kafka > Issue Type: Test > Components: core >Reporter: Bill Bejeck >Assignee: Rajini Sivaram >Priority: Major > Labels: flaky-test, tests > Fix For: 2.5.0 > > > Failed in > [https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/26571/testReport/junit/kafka.api/SaslGssapiSslEndToEndAuthorizationTest/testNoConsumeWithoutDescribeAclViaSubscribe/] > > {noformat} > Error Messageorg.apache.kafka.common.errors.TopicAuthorizationException: Not > authorized to access topics: > [topic2]Stacktraceorg.apache.kafka.common.errors.TopicAuthorizationException: > Not authorized to access topics: [topic2] > Standard OutputAdding ACLs for resource > `ResourcePattern(resourceType=CLUSTER, name=kafka-cluster, > patternType=LITERAL)`: > (principal=User:kafka, host=*, operation=CLUSTER_ACTION, > permissionType=ALLOW) > Current ACLs for resource `Cluster:LITERAL:kafka-cluster`: > User:kafka has Allow permission for operations: ClusterAction from > hosts: * > Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=*, > patternType=LITERAL)`: > (principal=User:kafka, host=*, operation=READ, permissionType=ALLOW) > Current ACLs for resource `Topic:LITERAL:*`: > User:kafka has Allow permission for operations: Read from hosts: * > Debug is true storeKey true useTicketCache false useKeyTab true doNotPrompt > false ticketCache is null isInitiator true KeyTab is > /tmp/kafka6494439724844851846.tmp refreshKrb5Config is false principal is > kafka/localh...@example.com tryFirstPass is false useFirstPass is false > storePass is false clearPass is false > principal is kafka/localh...@example.com > Will use keytab > Commit Succeeded > [2019-11-13 04:43:16,187] ERROR [ReplicaFetcher replicaId=1, leaderId=0, > fetcherId=0] Error for partition __consumer_offsets-0 at offset 0 > (kafka.server.ReplicaFetcherThread:76) > org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server > does not host this topic-partition. > [2019-11-13 04:43:16,191] ERROR [ReplicaFetcher replicaId=2, leaderId=0, > fetcherId=0] Error for partition __consumer_offsets-0 at offset 0 > (kafka.server.ReplicaFetcherThread:76) > org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server > does not host this topic-partition. > [2019-11-13 04:43:16,384] ERROR [ReplicaFetcher replicaId=1, leaderId=2, > fetcherId=0] Error for partition e2etopic-0 at offset 0 > (kafka.server.ReplicaFetcherThread:76) > org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server > does not host this topic-partition. > [2019-11-13 04:43:16,384] ERROR [ReplicaFetcher replicaId=0, leaderId=2, > fetcherId=0] Error for partition e2etopic-0 at offset 0 > (kafka.server.ReplicaFetcherThread:76) > org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server > does not host this topic-partition. > Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=e2etopic, > patternType=LITERAL)`: > (principal=User:client, host=*, operation=WRITE, permissionType=ALLOW) > (principal=User:client, host=*, operation=DESCRIBE, > permissionType=ALLOW) > (principal=User:client, host=*, operation=CREATE, permissionType=ALLOW) > Current ACLs for resource `Topic:LITERAL:e2etopic`: > User:client has Allow permission for operations: Describe from hosts: * > User:client has Allow permission for operations: Write from hosts: * > User:client has Allow permission for operations: Create from hosts: * > Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=e2etopic, > patternType=LITERAL)`: > (principal=User:client, host=*, operation=READ, permissionType=ALLOW) > (principal=User:client, host=*, operation=DESCRIBE, > permissionT
[jira] [Commented] (KAFKA-6266) Kafka 1.0.0 : Repeated occurrence of WARN Resetting first dirty offset of __consumer_offsets-xx to log start offset 203569 since the checkpointed offset 120955 is inval
[ https://issues.apache.org/jira/browse/KAFKA-6266?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034701#comment-17034701 ] ASF GitHub Bot commented on KAFKA-6266: --- splett2 commented on pull request #8089: KAFKA-6266: Repeated occurrence of WARN Resetting first dirty offset URL: https://github.com/apache/kafka/pull/8089 Previously, checkpointed offsets for a log were only updated if the log was chosen for cleaning once the cleaning job completes. This caused issues in cases where logs with invalid checkpointed offsets would repeatedly emit warnings if the log with an invalid cleaning checkpoint wasn't chosen for cleaning. Proposed fix is to update the checkpointed offset for logs with invalid checkpoints regardless of whether it gets chosen for cleaning. ### Committer Checklist (excluded from commit message) - [ ] Verify design and implementation - [ ] Verify test coverage and CI build status - [ ] Verify documentation (including upgrade notes) This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Kafka 1.0.0 : Repeated occurrence of WARN Resetting first dirty offset of > __consumer_offsets-xx to log start offset 203569 since the checkpointed > offset 120955 is invalid. (kafka.log.LogCleanerManager$) > -- > > Key: KAFKA-6266 > URL: https://issues.apache.org/jira/browse/KAFKA-6266 > Project: Kafka > Issue Type: Bug > Components: offset manager >Affects Versions: 1.0.0, 1.0.1 > Environment: CentOS 7, Apache kafka_2.12-1.0.0 >Reporter: VinayKumar >Assignee: David Mao >Priority: Major > Fix For: 2.5.0, 2.4.1 > > > I upgraded Kafka from 0.10.2.1 to 1.0.0 version. From then, I see the below > warnings in the log. > I'm seeing these continuously in the log, and want these to be fixed- so > that they wont repeat. Can someone please help me in fixing the below > warnings. > {code} > WARN Resetting first dirty offset of __consumer_offsets-17 to log start > offset 3346 since the checkpointed offset 3332 is invalid. > (kafka.log.LogCleanerManager$) > WARN Resetting first dirty offset of __consumer_offsets-23 to log start > offset 4 since the checkpointed offset 1 is invalid. > (kafka.log.LogCleanerManager$) > WARN Resetting first dirty offset of __consumer_offsets-19 to log start > offset 203569 since the checkpointed offset 120955 is invalid. > (kafka.log.LogCleanerManager$) > WARN Resetting first dirty offset of __consumer_offsets-35 to log start > offset 16957 since the checkpointed offset 7 is invalid. > (kafka.log.LogCleanerManager$) > {code} -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-9537) Abstract transformations in configurations cause unfriendly error message.
[ https://issues.apache.org/jira/browse/KAFKA-9537?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034713#comment-17034713 ] ASF GitHub Bot commented on KAFKA-9537: --- jcustenborder commented on pull request #8090: KAFKA-9537 - Cleanup error messages for abstract transformations URL: https://github.com/apache/kafka/pull/8090 Added check if the transformation is abstract. If so throw an error message with guidance for the user. *More detailed description of your change, if necessary. The PR title and PR message become the squashed commit message, so use a separate comment to ping reviewers.* *Summary of testing strategy (including rationale) for the feature or bug fix. Unit and/or integration tests are expected for any behaviour change and system tests should be considered for larger changes.* ### Committer Checklist (excluded from commit message) - [ ] Verify design and implementation - [ ] Verify test coverage and CI build status - [ ] Verify documentation (including upgrade notes) This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Abstract transformations in configurations cause unfriendly error message. > -- > > Key: KAFKA-9537 > URL: https://issues.apache.org/jira/browse/KAFKA-9537 > Project: Kafka > Issue Type: Bug > Components: KafkaConnect >Affects Versions: 2.4.0 >Reporter: Jeremy Custenborder >Assignee: Jeremy Custenborder >Priority: Minor > > I was working with a coworker who had a bash script posting a config to > connect with > {code:java}org.apache.kafka.connect.transforms.ExtractField.$Key{code} in the > script. Bash removed the $Key because it wasn't escaped properly. > {code:java} > org.apache.kafka.connect.transforms.ExtractField.{code} > is made it to the rest interface. A Class was create for the abstract > implementation of ExtractField and passed to getConfigDefFromTransformation. > It tried to call newInstance which threw an exception. The following gets > returned via the rest interface. > {code} > { > "error_code": 400, > "message": "Connector configuration is invalid and contains the following 1 > error(s):\nInvalid value class > org.apache.kafka.connect.transforms.ExtractField for configuration > transforms.extractString.type: Error getting config definition from > Transformation: null\nYou can also find the above list of errors at the > endpoint `/{connectorType}/config/validate`" > } > {code} > It would be a much better user experience if we returned something like > {code} > { > "error_code": 400, > "message": "Connector configuration is invalid and contains the following 1 > error(s):\nInvalid value class > org.apache.kafka.connect.transforms.ExtractField for configuration > transforms.extractString.type: Error getting config definition from > Transformation: Transformation is abstract and cannot be created.\nYou can > also find the above list of errors at the endpoint > `/{connectorType}/config/validate`" > } > {code} > or > {code} > { > "error_code": 400, > "message": "Connector configuration is invalid and contains the following 1 > error(s):\nInvalid value class > org.apache.kafka.connect.transforms.ExtractField for configuration > transforms.extractString.type: Error getting config definition from > Transformation: Transformation is abstract and cannot be created. Did you > mean ExtractField$Key, ExtractField$Value?\nYou can also find the above list > of errors at the endpoint `/{connectorType}/config/validate`" > } > {code} -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-8037) KTable restore may load bad data
[ https://issues.apache.org/jira/browse/KAFKA-8037?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034746#comment-17034746 ] Sophie Blee-Goldman commented on KAFKA-8037: [~pkleindl] [~vvcephei] [~mjsax] [~guozhang] I have kind of a different approach that might be a good compromise. We want to satisfy all of the following three goals as much as possible: # reuse the input topic as the changelog to avoid replicating all that data # make restoration as fast as possible by copying plain bytes with deserializing # don't load bad data, or copy bytes during restoration that weren't copied during normal processing As this ticket points out, we currently sacrifice 3) for the sake of 1) and 2) with the workaround being to give up 1) by turning off optimization. But what if we instead created a new type of internal topic that's essentially an "inverse-changelog" – rather than sending every record that goes to the store to this inverse-changelog, we send only the records that _don't_ get put into the store? In fact we don't even store the entire record, just the key bytes with a null value. Then once we've restored from the input-topic-changelog, we read from the inverse-changelog and any bad records get deleted without ever having to deserialize or store the value twice. My one concern is over the tracking/handling of local stream time: if we load the bad data during the initial restoration, it might bump up the stream time when it shouldn't have and potentially cause older, valid records to get dropped. If that's a real concern, then this approach would be blocked on KAFKA-9368 – however, I'm not confident that corrupted records don't currently bump the stream-time even during normal processing, and also not sure what kind of guarantees we should or do make w.r.t deserialization exceptions. Thoughts? > KTable restore may load bad data > > > Key: KAFKA-8037 > URL: https://issues.apache.org/jira/browse/KAFKA-8037 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: Matthias J. Sax >Priority: Minor > Labels: pull-request-available > > If an input topic contains bad data, users can specify a > `deserialization.exception.handler` to drop corrupted records on read. > However, this mechanism may be by-passed on restore. Assume a > `builder.table()` call reads and drops a corrupted record. If the table state > is lost and restored from the changelog topic, the corrupted record may be > copied into the store, because on restore plain bytes are copied. > If the KTable is used in a join, an internal `store.get()` call to lookup the > record would fail with a deserialization exception if the value part cannot > be deserialized. > GlobalKTables are affected, too (cf. KAFKA-7663 that may allow a fix for > GlobalKTable case). It's unclear to me atm, how this issue could be addressed > for KTables though. > Note, that user state stores are not affected, because they always have a > dedicated changelog topic (and don't reuse an input topic) and thus the > corrupted record would not be written into the changelog. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (KAFKA-9535) Metadata not updated when consumer encounters FENCED_LEADER_EPOCH
[ https://issues.apache.org/jira/browse/KAFKA-9535?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Boyang Chen updated KAFKA-9535: --- Description: Inside the consumer Fetcher's handling of ListOffsetResponse, if we hit `FENCED_LEADER_EPOCH` on partition level, the client will blindly retry without refreshing the metadata, creating a stuck state as the local leader epoch never gets updated and constantly fails the broker check. The solution is to trigger metadata update upon receiving retriable errors, before we kick off another offset list. was:Inside the consumer Fetcher's handling of ListOffsetResponse, if we hit `FENCED_LEADER_EPOCH` on partition level, the client will blindly retry without refreshing the metadata, creating a stuck state as the local leader epoch never gets updated and constantly fails the broker check. > Metadata not updated when consumer encounters FENCED_LEADER_EPOCH > - > > Key: KAFKA-9535 > URL: https://issues.apache.org/jira/browse/KAFKA-9535 > Project: Kafka > Issue Type: Bug >Affects Versions: 2.3.0, 2.4.0, 2.5.0 >Reporter: Boyang Chen >Priority: Major > > Inside the consumer Fetcher's handling of ListOffsetResponse, if we hit > `FENCED_LEADER_EPOCH` on partition level, the client will blindly retry > without refreshing the metadata, creating a stuck state as the local leader > epoch never gets updated and constantly fails the broker check. > The solution is to trigger metadata update upon receiving retriable errors, > before we kick off another offset list. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Assigned] (KAFKA-9535) Metadata not updated when consumer encounters FENCED_LEADER_EPOCH
[ https://issues.apache.org/jira/browse/KAFKA-9535?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Boyang Chen reassigned KAFKA-9535: -- Assignee: Boyang Chen > Metadata not updated when consumer encounters FENCED_LEADER_EPOCH > - > > Key: KAFKA-9535 > URL: https://issues.apache.org/jira/browse/KAFKA-9535 > Project: Kafka > Issue Type: Bug >Affects Versions: 2.3.0, 2.4.0, 2.5.0 >Reporter: Boyang Chen >Assignee: Boyang Chen >Priority: Major > > Inside the consumer Fetcher's handling of ListOffsetResponse, if we hit > `FENCED_LEADER_EPOCH` on partition level, the client will blindly retry > without refreshing the metadata, creating a stuck state as the local leader > epoch never gets updated and constantly fails the broker check. > The solution is to trigger metadata update upon receiving retriable errors, > before we kick off another offset list. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Resolved] (KAFKA-8616) Replace ApiVersionsRequest request/response with automated protocol
[ https://issues.apache.org/jira/browse/KAFKA-8616?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Boyang Chen resolved KAFKA-8616. Resolution: Duplicate > Replace ApiVersionsRequest request/response with automated protocol > --- > > Key: KAFKA-8616 > URL: https://issues.apache.org/jira/browse/KAFKA-8616 > Project: Kafka > Issue Type: Sub-task >Reporter: Boyang Chen >Assignee: Boyang Chen >Priority: Major > -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (KAFKA-9538) Flaky Test `testResetOffsetsExportImportPlan`
Jason Gustafson created KAFKA-9538: -- Summary: Flaky Test `testResetOffsetsExportImportPlan` Key: KAFKA-9538 URL: https://issues.apache.org/jira/browse/KAFKA-9538 Project: Kafka Issue Type: Bug Reporter: Jason Gustafson Assignee: Jason Gustafson {code} 19:44:41 19:44:41 kafka.admin.ResetConsumerGroupOffsetTest > testResetOffsetsExportImportPlan FAILED 19:44:41 java.lang.AssertionError: expected: 2, bar2-1 -> 2)> but was: 19:44:41 at org.junit.Assert.fail(Assert.java:89) 19:44:41 at org.junit.Assert.failNotEquals(Assert.java:835) 19:44:41 at org.junit.Assert.assertEquals(Assert.java:120) 19:44:41 at org.junit.Assert.assertEquals(Assert.java:146) 19:44:41 at kafka.admin.ResetConsumerGroupOffsetTest.testResetOffsetsExportImportPlan(ResetConsumerGroupOffsetTest.scala:429) {code} -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Resolved] (KAFKA-8211) Flaky Test: ResetConsumerGroupOffsetTest.testResetOffsetsExportImportPlan
[ https://issues.apache.org/jira/browse/KAFKA-8211?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jason Gustafson resolved KAFKA-8211. Resolution: Fixed Closing this one since new failures do not match the stacktrace. I've opened https://issues.apache.org/jira/browse/KAFKA-9538 instead. > Flaky Test: ResetConsumerGroupOffsetTest.testResetOffsetsExportImportPlan > - > > Key: KAFKA-8211 > URL: https://issues.apache.org/jira/browse/KAFKA-8211 > Project: Kafka > Issue Type: Bug > Components: admin, clients, unit tests >Affects Versions: 2.3.0 >Reporter: Bill Bejeck >Assignee: huxihx >Priority: Major > Fix For: 2.5.0 > > > Failed in build [https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/20778/] > > {noformat} > Error Message > java.lang.AssertionError: Expected that consumer group has consumed all > messages from topic/partition. > Stacktrace > java.lang.AssertionError: Expected that consumer group has consumed all > messages from topic/partition. > at kafka.utils.TestUtils$.fail(TestUtils.scala:381) > at kafka.utils.TestUtils$.waitUntilTrue(TestUtils.scala:791) > at > kafka.admin.ResetConsumerGroupOffsetTest.awaitConsumerProgress(ResetConsumerGroupOffsetTest.scala:364) > at > kafka.admin.ResetConsumerGroupOffsetTest.produceConsumeAndShutdown(ResetConsumerGroupOffsetTest.scala:359) > at > kafka.admin.ResetConsumerGroupOffsetTest.testResetOffsetsExportImportPlan(ResetConsumerGroupOffsetTest.scala:323) > at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) > at > sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) > at > sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) > at java.lang.reflect.Method.invoke(Method.java:498) > 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.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26) > at > org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27) > 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.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26) > at > org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27) > 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 sun.reflect.GeneratedMethodAccessor11.invoke(Unknown Source) > at > sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) > at java.lang.reflect.Method.invoke(Method.java:498) > at > org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:35) > at > org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24) > at > org.gradle.internal.dispatch.ContextClassLoaderDispatch.dispatch(ContextClassLoaderDispatch.java:32) > at > org.gradle.internal.dispatch.ProxyDis
[jira] [Assigned] (KAFKA-9538) Flaky Test `testResetOffsetsExportImportPlan`
[ https://issues.apache.org/jira/browse/KAFKA-9538?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jason Gustafson reassigned KAFKA-9538: -- Assignee: huxihx (was: Jason Gustafson) > Flaky Test `testResetOffsetsExportImportPlan` > - > > Key: KAFKA-9538 > URL: https://issues.apache.org/jira/browse/KAFKA-9538 > Project: Kafka > Issue Type: Bug >Reporter: Jason Gustafson >Assignee: huxihx >Priority: Major > > {code} > 19:44:41 > 19:44:41 kafka.admin.ResetConsumerGroupOffsetTest > > testResetOffsetsExportImportPlan FAILED > 19:44:41 java.lang.AssertionError: expected: 2, bar2-1 -> > 2)> but was: > 19:44:41 at org.junit.Assert.fail(Assert.java:89) > 19:44:41 at org.junit.Assert.failNotEquals(Assert.java:835) > 19:44:41 at org.junit.Assert.assertEquals(Assert.java:120) > 19:44:41 at org.junit.Assert.assertEquals(Assert.java:146) > 19:44:41 at > kafka.admin.ResetConsumerGroupOffsetTest.testResetOffsetsExportImportPlan(ResetConsumerGroupOffsetTest.scala:429) > {code} -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-8940) Flaky Test SmokeTestDriverIntegrationTest.shouldWorkWithRebalance
[ https://issues.apache.org/jira/browse/KAFKA-8940?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034766#comment-17034766 ] Sophie Blee-Goldman commented on KAFKA-8940: Failed again for me (locally): {code:java} java.lang.AssertionError: verifying tagg fail: key=1164 tagg=[ConsumerRecord(topic = tagg, partition = 0, leaderEpoch = 0, offset = 33, CreateTime = 1581392865550, serialized key size = 4, serialized value size = 8, headers = RecordHeaders(headers = [], isReadOnly = false), key = 1164, value = 1)] expected=0 taggEvents: [ConsumerRecord(topic = tagg, partition = 0, leaderEpoch = 0, offset = 33, CreateTime = 1581392865550, serialized key size = 4, serialized value size = 8, headers = RecordHeaders(headers = [], isReadOnly = false), key = 1164, value = 1)] verifying suppressed min-suppressed verifying min-suppressed with 10 keys verifying suppressed sws-suppressed verifying min with 10 keys verifying max with 10 keys verifying dif with 10 keys verifying sum with 10 keys sum fail: key=7-1006 actual=539274 expected=506500{code} > Flaky Test SmokeTestDriverIntegrationTest.shouldWorkWithRebalance > - > > Key: KAFKA-8940 > URL: https://issues.apache.org/jira/browse/KAFKA-8940 > Project: Kafka > Issue Type: Bug > Components: streams, unit tests >Reporter: Guozhang Wang >Assignee: Guozhang Wang >Priority: Major > Labels: flaky-test > Fix For: 2.5.0 > > > I lost the screen shot unfortunately... it reports the set of expected > records does not match the received records. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-7052) ExtractField SMT throws NPE - needs clearer error message
[ https://issues.apache.org/jira/browse/KAFKA-7052?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034777#comment-17034777 ] ASF GitHub Bot commented on KAFKA-7052: --- rhauch commented on pull request #8059: KAFKA-7052 Avoiding NPE in ExtractField SMT in case of non-existent fields URL: https://github.com/apache/kafka/pull/8059 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > ExtractField SMT throws NPE - needs clearer error message > - > > Key: KAFKA-7052 > URL: https://issues.apache.org/jira/browse/KAFKA-7052 > Project: Kafka > Issue Type: Improvement > Components: KafkaConnect >Reporter: Robin Moffatt >Priority: Major > > With the following Single Message Transform: > {code:java} > "transforms.ExtractId.type":"org.apache.kafka.connect.transforms.ExtractField$Key", > "transforms.ExtractId.field":"id"{code} > Kafka Connect errors with : > {code:java} > java.lang.NullPointerException > at > org.apache.kafka.connect.transforms.ExtractField.apply(ExtractField.java:61) > at > org.apache.kafka.connect.runtime.TransformationChain.apply(TransformationChain.java:38){code} > There should be a better error message here, identifying the reason for the > NPE. > Version: Confluent Platform 4.1.1 -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (KAFKA-9417) Integration test for new EOS model with vanilla Producer and Consumer
[ https://issues.apache.org/jira/browse/KAFKA-9417?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Boyang Chen updated KAFKA-9417: --- Description: We would like to extend the `TransactionMessageCopier` to use the new subscription mode consumer and do a system test based off that in order to verify the new semantic actually works. We also want to make sure the backward compatibility is maintained by using group metadata API in existing tests as well. A minor public change is also included within this PR by setting `transaction.abort.timed.out.transaction.cleanup.interval.ms` default to 1 ms (10 seconds) on broker side was: We would like to extend the `TransactionMessageCopier` to use the new subscription mode consumer and do a system test based off that in order to verify the new semantic actually works. We also want to make sure the backward compatibility is maintained by using group metadata API in existing tests as well. > Integration test for new EOS model with vanilla Producer and Consumer > - > > Key: KAFKA-9417 > URL: https://issues.apache.org/jira/browse/KAFKA-9417 > Project: Kafka > Issue Type: Sub-task > Components: streams >Reporter: Boyang Chen >Assignee: Boyang Chen >Priority: Major > > We would like to extend the `TransactionMessageCopier` to use the new > subscription mode consumer and do a system test based off that in order to > verify the new semantic actually works. > We also want to make sure the backward compatibility is maintained by using > group metadata API in existing tests as well. > > A minor public change is also included within this PR by setting > `transaction.abort.timed.out.transaction.cleanup.interval.ms` default to > 1 ms (10 seconds) on broker side -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Resolved] (KAFKA-7052) ExtractField SMT throws NPE - needs clearer error message
[ https://issues.apache.org/jira/browse/KAFKA-7052?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Randall Hauch resolved KAFKA-7052. -- Fix Version/s: 2.4.1 2.3.2 2.5.0 2.2.3 Reviewer: Randall Hauch Resolution: Fixed Merged to the `trunk`, `2.5`, `2.4`, `2.3` and `2.2` branches. > ExtractField SMT throws NPE - needs clearer error message > - > > Key: KAFKA-7052 > URL: https://issues.apache.org/jira/browse/KAFKA-7052 > Project: Kafka > Issue Type: Improvement > Components: KafkaConnect >Reporter: Robin Moffatt >Priority: Major > Fix For: 2.2.3, 2.5.0, 2.3.2, 2.4.1 > > > With the following Single Message Transform: > {code:java} > "transforms.ExtractId.type":"org.apache.kafka.connect.transforms.ExtractField$Key", > "transforms.ExtractId.field":"id"{code} > Kafka Connect errors with : > {code:java} > java.lang.NullPointerException > at > org.apache.kafka.connect.transforms.ExtractField.apply(ExtractField.java:61) > at > org.apache.kafka.connect.runtime.TransformationChain.apply(TransformationChain.java:38){code} > There should be a better error message here, identifying the reason for the > NPE. > Version: Confluent Platform 4.1.1 -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (KAFKA-9539) Add leader epoch in StopReplicaRequest
David Jacot created KAFKA-9539: -- Summary: Add leader epoch in StopReplicaRequest Key: KAFKA-9539 URL: https://issues.apache.org/jira/browse/KAFKA-9539 Project: Kafka Issue Type: Improvement Reporter: David Jacot Assignee: David Jacot Unlike the LeaderAndIsrRequest, the StopReplicaRequest does not include the leader epoch which makes it vulnerable to reordering. This KIP proposes to add the leader epoch for each partition in the StopReplicaRequest and the broker will verify the epoch before proceeding with the StopReplicaRequest. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-9479) Describe consumer group --all-groups shows header for each entry
[ https://issues.apache.org/jira/browse/KAFKA-9479?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034832#comment-17034832 ] Jason Gustafson commented on KAFKA-9479: [~jeffkbkim] If you have not already started on this, perhaps we could let [~ve...@roeim.net] submit the patch? > Describe consumer group --all-groups shows header for each entry > > > Key: KAFKA-9479 > URL: https://issues.apache.org/jira/browse/KAFKA-9479 > Project: Kafka > Issue Type: Improvement >Reporter: Jason Gustafson >Assignee: Jeff Kim >Priority: Major > Labels: newbie > > When using `bin/kafka-consumer-groups.sh --describe --state --all-groups`, we > print output like the following: > {code} > GROUP COORDINATOR (ID) > ASSIGNMENT-STRATEGY STATE #MEMBERS > group1 localhost:9092 (3) rangeStable 1 > > > GROUP COORDINATOR (ID) > ASSIGNMENT-STRATEGY STATE #MEMBERS > group2 localhost:9092 (3) rangeStable 1 > > > {code} > It would be nice if we did not show the header for every entry. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-9013) Flaky Test MirrorConnectorsIntegrationTest#testReplication
[ https://issues.apache.org/jira/browse/KAFKA-9013?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034850#comment-17034850 ] Ryanne Dolan commented on KAFKA-9013: - I'm seeing loglines like `Found 11 topic-partitions on primary. 0 are new. 1 were removed. Previously had 11.` That doesn't make sense. It looks like the log message is wrong actually (it shows the same count "11" twice -- confirmed as wrong in the code), but more importantly it means the tasks are being re-configured repeatedly, which might account for the flaky test results. [~ecomar] can you take a look at that? Do we know why MirrorSourceConnector would think that upstreamTargetTopicPartitions - knownSourceTopicPartitions is non-empty? > Flaky Test MirrorConnectorsIntegrationTest#testReplication > -- > > Key: KAFKA-9013 > URL: https://issues.apache.org/jira/browse/KAFKA-9013 > Project: Kafka > Issue Type: Bug > Components: KafkaConnect >Reporter: Bruno Cadonna >Priority: Major > Labels: flaky-test > > h1. Stacktrace: > {code:java} > java.lang.AssertionError: Condition not met within timeout 2. Offsets not > translated downstream to primary cluster. > at org.apache.kafka.test.TestUtils.waitForCondition(TestUtils.java:377) > at org.apache.kafka.test.TestUtils.waitForCondition(TestUtils.java:354) > at > org.apache.kafka.connect.mirror.MirrorConnectorsIntegrationTest.testReplication(MirrorConnectorsIntegrationTest.java:239) > {code} > h1. Standard Error > {code} > Standard Error > Oct 09, 2019 11:32:00 PM org.glassfish.jersey.internal.inject.Providers > checkProviderRuntime > WARNING: A provider > org.apache.kafka.connect.runtime.rest.resources.ConnectorPluginsResource > registered in SERVER runtime does not implement any provider interfaces > applicable in the SERVER runtime. Due to constraint configuration problems > the provider > org.apache.kafka.connect.runtime.rest.resources.ConnectorPluginsResource will > be ignored. > Oct 09, 2019 11:32:00 PM org.glassfish.jersey.internal.inject.Providers > checkProviderRuntime > WARNING: A provider > org.apache.kafka.connect.runtime.rest.resources.LoggingResource registered in > SERVER runtime does not implement any provider interfaces applicable in the > SERVER runtime. Due to constraint configuration problems the provider > org.apache.kafka.connect.runtime.rest.resources.LoggingResource will be > ignored. > Oct 09, 2019 11:32:00 PM org.glassfish.jersey.internal.inject.Providers > checkProviderRuntime > WARNING: A provider > org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource registered > in SERVER runtime does not implement any provider interfaces applicable in > the SERVER runtime. Due to constraint configuration problems the provider > org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource will be > ignored. > Oct 09, 2019 11:32:00 PM org.glassfish.jersey.internal.inject.Providers > checkProviderRuntime > WARNING: A provider > org.apache.kafka.connect.runtime.rest.resources.RootResource registered in > SERVER runtime does not implement any provider interfaces applicable in the > SERVER runtime. Due to constraint configuration problems the provider > org.apache.kafka.connect.runtime.rest.resources.RootResource will be ignored. > Oct 09, 2019 11:32:01 PM org.glassfish.jersey.internal.Errors logErrors > WARNING: The following warnings have been detected: WARNING: The > (sub)resource method listLoggers in > org.apache.kafka.connect.runtime.rest.resources.LoggingResource contains > empty path annotation. > WARNING: The (sub)resource method listConnectors in > org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource contains > empty path annotation. > WARNING: The (sub)resource method createConnector in > org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource contains > empty path annotation. > WARNING: The (sub)resource method listConnectorPlugins in > org.apache.kafka.connect.runtime.rest.resources.ConnectorPluginsResource > contains empty path annotation. > WARNING: The (sub)resource method serverInfo in > org.apache.kafka.connect.runtime.rest.resources.RootResource contains empty > path annotation. > Oct 09, 2019 11:32:02 PM org.glassfish.jersey.internal.inject.Providers > checkProviderRuntime > WARNING: A provider > org.apache.kafka.connect.runtime.rest.resources.ConnectorPluginsResource > registered in SERVER runtime does not implement any provider interfaces > applicable in the SERVER runtime. Due to constraint configuration problems > the provider > org.apache.kafka.connect.runtime.rest.resources.ConnectorPluginsResource will > be ignored. > Oct 09, 2019 11:32:02 PM org.glassfish.jersey.internal.inject.Providers > checkProviderRunt
[jira] [Commented] (KAFKA-9479) Describe consumer group --all-groups shows header for each entry
[ https://issues.apache.org/jira/browse/KAFKA-9479?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034881#comment-17034881 ] Jeff Kim commented on KAFKA-9479: - [~hachikuji] sounds good. [~ve...@roeim.net] could you submit the patch? > Describe consumer group --all-groups shows header for each entry > > > Key: KAFKA-9479 > URL: https://issues.apache.org/jira/browse/KAFKA-9479 > Project: Kafka > Issue Type: Improvement >Reporter: Jason Gustafson >Assignee: Jeff Kim >Priority: Major > Labels: newbie > > When using `bin/kafka-consumer-groups.sh --describe --state --all-groups`, we > print output like the following: > {code} > GROUP COORDINATOR (ID) > ASSIGNMENT-STRATEGY STATE #MEMBERS > group1 localhost:9092 (3) rangeStable 1 > > > GROUP COORDINATOR (ID) > ASSIGNMENT-STRATEGY STATE #MEMBERS > group2 localhost:9092 (3) rangeStable 1 > > > {code} > It would be nice if we did not show the header for every entry. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-6607) Consumer Client and Kafka Streams lag not zero when input topic transactional
[ https://issues.apache.org/jira/browse/KAFKA-6607?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034882#comment-17034882 ] ASF GitHub Bot commented on KAFKA-6607: --- mjsax commented on pull request #8091: KAFKA-6607: Commit correct offsets for transactional input data URL: https://github.com/apache/kafka/pull/8091 Port of #8040 to `trunk` -- the only real difference is the missing check of we are in pending shutdown and a task is not found -- in `trunk` we always create all new task hence this check is not necessary. \cc @guozhangwang This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Consumer Client and Kafka Streams lag not zero when input topic transactional > - > > Key: KAFKA-6607 > URL: https://issues.apache.org/jira/browse/KAFKA-6607 > Project: Kafka > Issue Type: Improvement > Components: clients, streams >Affects Versions: 1.0.0 >Reporter: Matthias J. Sax >Assignee: Matthias J. Sax >Priority: Minor > > When an input topic for a Consumer or Kafka Streams application is written > using transaction, the client does not commit "endOffset" but "endOffset - 1" > (or to be more precise, "lastProcessedMessageOffset + 1") if it reaches the > end of topic. The reason is the commit marker that is the last "message" in > the topic; Streams commit "offset of last processed message plus 1" and does > not take commit markers into account. > This is not a correctness issue, but when one inspect the consumer lag via > {{bin/kafka-consumer.group.sh}} the lag is show as 1 instead of 0 – what is > correct from consumer-group tool point of view. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-6607) Consumer Client and Kafka Streams lag not zero when input topic transactional
[ https://issues.apache.org/jira/browse/KAFKA-6607?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034884#comment-17034884 ] ASF GitHub Bot commented on KAFKA-6607: --- mjsax commented on pull request #8040: KAFKA-6607: Commit correct offsets for transactional input data URL: https://github.com/apache/kafka/pull/8040 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Consumer Client and Kafka Streams lag not zero when input topic transactional > - > > Key: KAFKA-6607 > URL: https://issues.apache.org/jira/browse/KAFKA-6607 > Project: Kafka > Issue Type: Improvement > Components: clients, streams >Affects Versions: 1.0.0 >Reporter: Matthias J. Sax >Assignee: Matthias J. Sax >Priority: Minor > > When an input topic for a Consumer or Kafka Streams application is written > using transaction, the client does not commit "endOffset" but "endOffset - 1" > (or to be more precise, "lastProcessedMessageOffset + 1") if it reaches the > end of topic. The reason is the commit marker that is the last "message" in > the topic; Streams commit "offset of last processed message plus 1" and does > not take commit markers into account. > This is not a correctness issue, but when one inspect the consumer lag via > {{bin/kafka-consumer.group.sh}} the lag is show as 1 instead of 0 – what is > correct from consumer-group tool point of view. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-9530) Flaky Test kafka.admin.DescribeConsumerGroupTest.testDescribeGroupWithShortInitializationTimeout
[ https://issues.apache.org/jira/browse/KAFKA-9530?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034885#comment-17034885 ] Matthias J. Sax commented on KAFKA-9530: [https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/630/testReport/junit/kafka.admin/DescribeConsumerGroupTest/testDescribeGroupWithShortInitializationTimeout/] > Flaky Test > kafka.admin.DescribeConsumerGroupTest.testDescribeGroupWithShortInitializationTimeout > > > Key: KAFKA-9530 > URL: https://issues.apache.org/jira/browse/KAFKA-9530 > Project: Kafka > Issue Type: Test > Components: core >Reporter: Bill Bejeck >Priority: Major > Labels: flaky-test, test > > [https://builds.apache.org/job/kafka-pr-jdk11-scala2.13/4570/testReport/junit/kafka.admin/DescribeConsumerGroupTest/testDescribeGroupWithShortInitializationTimeout/] > > {noformat} > Error Messagejava.lang.AssertionError: assertion > failedStacktracejava.lang.AssertionError: assertion failed > at scala.Predef$.assert(Predef.scala:267) > at > kafka.admin.DescribeConsumerGroupTest.testDescribeGroupWithShortInitializationTimeout(DescribeConsumerGroupTest.scala:585) > 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.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26) > at > org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27) > at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306) > at > org.junit.runners.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100) > at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:366) > 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:331) > at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79) > at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:329) > at org.junit.runners.ParentRunner.access$100(ParentRunner.java:66) > at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:293) > at > org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26) > at > org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27) > at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306) > at org.junit.runners.ParentRunner.run(ParentRunner.java:413) > 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 jdk.internal.reflect.GeneratedMethodAccessor28.invoke(Unknown Source) > at > java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) > at java.base/java.lang.reflect.Method.invoke(Method.java:566) > at > org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:36) > at > org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24) > at > org.gradle.internal.dispatch.ContextClassLoaderDispatch.dispatch(ContextClassLoaderDispatch.java:33) > at > org.gradle.internal.dispatch.ProxyDispatchAdapter$DispatchingInvocationHandler.invoke(ProxyDispatchAdapter.java:94) > at com.sun.proxy.$Proxy2.processTestClass(Unknown Source) > at > org.gradle.api.internal.tasks.testing.worker.Te
[jira] [Commented] (KAFKA-8110) Flaky Test DescribeConsumerGroupTest#testDescribeMembersWithConsumersWithoutAssignedPartitions
[ https://issues.apache.org/jira/browse/KAFKA-8110?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034887#comment-17034887 ] Matthias J. Sax commented on KAFKA-8110: [https://builds.apache.org/job/kafka-pr-jdk11-scala2.13/4643/testReport/junit/kafka.admin/DescribeConsumerGroupTest/testDescribeGroupMembersWithShortInitializationTimeout/] > Flaky Test > DescribeConsumerGroupTest#testDescribeMembersWithConsumersWithoutAssignedPartitions > -- > > Key: KAFKA-8110 > URL: https://issues.apache.org/jira/browse/KAFKA-8110 > Project: Kafka > Issue Type: Bug > Components: core, unit tests >Affects Versions: 2.2.0 >Reporter: Matthias J. Sax >Priority: Critical > Labels: flaky-test > Fix For: 2.5.0 > > > [https://jenkins.confluent.io/job/apache-kafka-test/job/2.2/67/testReport/junit/kafka.admin/DescribeConsumerGroupTest/testDescribeMembersWithConsumersWithoutAssignedPartitions/] > {quote}java.lang.AssertionError: Partition [__consumer_offsets,0] metadata > not propagated after 15000 ms at > kafka.utils.TestUtils$.fail(TestUtils.scala:381) at > kafka.utils.TestUtils$.waitUntilTrue(TestUtils.scala:791) at > kafka.utils.TestUtils$.waitUntilMetadataIsPropagated(TestUtils.scala:880) at > kafka.utils.TestUtils$.$anonfun$createTopic$3(TestUtils.scala:318) at > kafka.utils.TestUtils$.$anonfun$createTopic$3$adapted(TestUtils.scala:317) at > scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:237) at > scala.collection.immutable.Range.foreach(Range.scala:158) at > scala.collection.TraversableLike.map(TraversableLike.scala:237) at > scala.collection.TraversableLike.map$(TraversableLike.scala:230) at > scala.collection.AbstractTraversable.map(Traversable.scala:108) at > kafka.utils.TestUtils$.createTopic(TestUtils.scala:317) at > kafka.utils.TestUtils$.createOffsetsTopic(TestUtils.scala:375) at > kafka.admin.DescribeConsumerGroupTest.testDescribeMembersWithConsumersWithoutAssignedPartitions(DescribeConsumerGroupTest.scala:372){quote} > STDOUT > {quote}[2019-03-14 20:01:52,347] WARN Ignoring unexpected runtime exception > (org.apache.zookeeper.server.NIOServerCnxnFactory:236) > java.nio.channels.CancelledKeyException at > sun.nio.ch.SelectionKeyImpl.ensureValid(SelectionKeyImpl.java:73) at > sun.nio.ch.SelectionKeyImpl.readyOps(SelectionKeyImpl.java:87) at > org.apache.zookeeper.server.NIOServerCnxnFactory.run(NIOServerCnxnFactory.java:205) > at java.lang.Thread.run(Thread.java:748) TOPIC PARTITION CURRENT-OFFSET > LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID foo 0 0 0 0 - - - TOPIC > PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID foo 0 > 0 0 0 - - - COORDINATOR (ID) ASSIGNMENT-STRATEGY STATE #MEMBERS > localhost:44669 (0){quote} -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Issue Comment Deleted] (KAFKA-9530) Flaky Test kafka.admin.DescribeConsumerGroupTest.testDescribeGroupWithShortInitializationTimeout
[ https://issues.apache.org/jira/browse/KAFKA-9530?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Matthias J. Sax updated KAFKA-9530: --- Comment: was deleted (was: [https://builds.apache.org/job/kafka-pr-jdk11-scala2.13/4643/testReport/junit/kafka.admin/DescribeConsumerGroupTest/testDescribeGroupMembersWithShortInitializationTimeout/]) > Flaky Test > kafka.admin.DescribeConsumerGroupTest.testDescribeGroupWithShortInitializationTimeout > > > Key: KAFKA-9530 > URL: https://issues.apache.org/jira/browse/KAFKA-9530 > Project: Kafka > Issue Type: Test > Components: core >Reporter: Bill Bejeck >Priority: Major > Labels: flaky-test, test > > [https://builds.apache.org/job/kafka-pr-jdk11-scala2.13/4570/testReport/junit/kafka.admin/DescribeConsumerGroupTest/testDescribeGroupWithShortInitializationTimeout/] > > {noformat} > Error Messagejava.lang.AssertionError: assertion > failedStacktracejava.lang.AssertionError: assertion failed > at scala.Predef$.assert(Predef.scala:267) > at > kafka.admin.DescribeConsumerGroupTest.testDescribeGroupWithShortInitializationTimeout(DescribeConsumerGroupTest.scala:585) > 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.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26) > at > org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27) > at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306) > at > org.junit.runners.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100) > at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:366) > 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:331) > at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79) > at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:329) > at org.junit.runners.ParentRunner.access$100(ParentRunner.java:66) > at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:293) > at > org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26) > at > org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27) > at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306) > at org.junit.runners.ParentRunner.run(ParentRunner.java:413) > 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 jdk.internal.reflect.GeneratedMethodAccessor28.invoke(Unknown Source) > at > java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) > at java.base/java.lang.reflect.Method.invoke(Method.java:566) > at > org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:36) > at > org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24) > at > org.gradle.internal.dispatch.ContextClassLoaderDispatch.dispatch(ContextClassLoaderDispatch.java:33) > at > org.gradle.internal.dispatch.ProxyDispatchAdapter$DispatchingInvocationHandler.invoke(ProxyDispatchAdapter.java:94) > at com.sun.proxy.$Proxy2.processTestClass(Unknown Source) > at > org.gradle.api.internal.tasks.testing.worker.TestWorker.proces
[jira] [Commented] (KAFKA-9530) Flaky Test kafka.admin.DescribeConsumerGroupTest.testDescribeGroupWithShortInitializationTimeout
[ https://issues.apache.org/jira/browse/KAFKA-9530?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034886#comment-17034886 ] Matthias J. Sax commented on KAFKA-9530: [https://builds.apache.org/job/kafka-pr-jdk11-scala2.13/4643/testReport/junit/kafka.admin/DescribeConsumerGroupTest/testDescribeGroupMembersWithShortInitializationTimeout/] > Flaky Test > kafka.admin.DescribeConsumerGroupTest.testDescribeGroupWithShortInitializationTimeout > > > Key: KAFKA-9530 > URL: https://issues.apache.org/jira/browse/KAFKA-9530 > Project: Kafka > Issue Type: Test > Components: core >Reporter: Bill Bejeck >Priority: Major > Labels: flaky-test, test > > [https://builds.apache.org/job/kafka-pr-jdk11-scala2.13/4570/testReport/junit/kafka.admin/DescribeConsumerGroupTest/testDescribeGroupWithShortInitializationTimeout/] > > {noformat} > Error Messagejava.lang.AssertionError: assertion > failedStacktracejava.lang.AssertionError: assertion failed > at scala.Predef$.assert(Predef.scala:267) > at > kafka.admin.DescribeConsumerGroupTest.testDescribeGroupWithShortInitializationTimeout(DescribeConsumerGroupTest.scala:585) > 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.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26) > at > org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27) > at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306) > at > org.junit.runners.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100) > at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:366) > 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:331) > at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79) > at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:329) > at org.junit.runners.ParentRunner.access$100(ParentRunner.java:66) > at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:293) > at > org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26) > at > org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27) > at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306) > at org.junit.runners.ParentRunner.run(ParentRunner.java:413) > 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 jdk.internal.reflect.GeneratedMethodAccessor28.invoke(Unknown Source) > at > java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) > at java.base/java.lang.reflect.Method.invoke(Method.java:566) > at > org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:36) > at > org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24) > at > org.gradle.internal.dispatch.ContextClassLoaderDispatch.dispatch(ContextClassLoaderDispatch.java:33) > at > org.gradle.internal.dispatch.ProxyDispatchAdapter$DispatchingInvocationHandler.invoke(ProxyDispatchAdapter.java:94) > at com.sun.proxy.$Proxy2.processTestClass(Unknown Source) > at > org.gradle.api.internal.tasks.testing.
[jira] [Commented] (KAFKA-9455) Consider using TreeMap for in-memory stores of Streams
[ https://issues.apache.org/jira/browse/KAFKA-9455?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034889#comment-17034889 ] Guozhang Wang commented on KAFKA-9455: -- >From the ReadOnlyWindowedStore interface: {code} V fetch(K key, long time);// single-point query WindowStoreIterator fetch(K key, Instant from, Instant to);// range-query KeyValueIterator, V> fetch(K from, K to, Instant fromTime, Instant toTime); // range-query KeyValueIterator, V> all(); // range-query KeyValueIterator, V> fetchAll(Instant from, Instant to); // range-query {code} The other APIs are deprecated. Hope this helps? > Consider using TreeMap for in-memory stores of Streams > -- > > Key: KAFKA-9455 > URL: https://issues.apache.org/jira/browse/KAFKA-9455 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: Guozhang Wang >Assignee: highluck >Priority: Major > Labels: newbie++ > > From [~ableegoldman]: It's worth noting that it might be a good idea to > switch to TreeMap for different reasons. Right now the ConcurrentSkipListMap > allows us to safely perform range queries without copying over the entire > keyset, but the performance on point queries seems to scale noticeably worse > with the number of unique keys. Point queries are used by aggregations while > range queries are used by windowed joins, but of course both are available > within the PAPI and for interactive queries so it's hard to say which we > should prefer. Maybe rather than make that tradeoff we should have one > version for efficient range queries (a "JoinWindowStore") and one for > efficient point queries ("AggWindowStore") - or something. I know we've had > similar thoughts for a different RocksDB store layout for Joins (although I > can't find that ticket anywhere..), it seems like the in-memory stores could > benefit from a special "Join" version as well cc/ Guozhang Wang > Here are some random thoughts: > 1. For kafka streams processing logic (i.e. without IQ), it's better to make > all processing logic relying on point queries rather than range queries. > Right now the only processor that use range queries are, as mentioned above, > windowed stream-stream joins. I think we should consider using a different > window implementation for this (and as a result also get rid of the > retainDuplicate flags) to refactor the windowed stream-stream join operation. > 2. With 1), range queries would only be exposed as IQ. Depending on its usage > frequency I think it makes lots of sense to optimize for single-point queries. > Of course, even without step 1) we should still consider using tree-map for > windowed in-memory stores to have a better scaling effect. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (KAFKA-9540) Application getting "Could not find the standby task 0_4 while closing it" error
Badai Aqrandista created KAFKA-9540: --- Summary: Application getting "Could not find the standby task 0_4 while closing it" error Key: KAFKA-9540 URL: https://issues.apache.org/jira/browse/KAFKA-9540 Project: Kafka Issue Type: Bug Components: streams Reporter: Badai Aqrandista Because of this the following line, there is a possibility that some standby tasks might not be created: https://github.com/apache/kafka/blob/2.4.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java#L436 Then causing this line to not adding the task to standby task list: https://github.com/apache/kafka/blob/2.4.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java#L299 But this line assumes that all standby tasks are to be created and add it to the standby list: https://github.com/apache/kafka/blob/2.4.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java#L168 This results in user getting this error message on the next PARTITION_ASSIGNMENT state: {noformat} Could not find the standby task 0_4 while closing it (org.apache.kafka.streams.processor.internals.AssignedStandbyTasks:74) {noformat} But the harm caused by this issue is minimal: No standby task for some partitions. And it is recreated on the next rebalance anyway. So, I suggest lowering this message to WARN. Or probably check to WARN when standby task could not be created. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-9483) Add Scala KStream#toTable to the Streams DSL
[ https://issues.apache.org/jira/browse/KAFKA-9483?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034892#comment-17034892 ] ASF GitHub Bot commented on KAFKA-9483: --- mjsax commented on pull request #8024: KAFKA-9483; Add Scala KStream#toTable to the Streams DSL URL: https://github.com/apache/kafka/pull/8024 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Add Scala KStream#toTable to the Streams DSL > > > Key: KAFKA-9483 > URL: https://issues.apache.org/jira/browse/KAFKA-9483 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: highluck >Assignee: highluck >Priority: Minor > > [KIP-523: Add KStream#toTable to the Streams > DSL|https://cwiki.apache.org/confluence/display/KAFKA/KIP-523%3A+Add+KStream%23toTable+to+the+Streams+DSL?src=jira] > > I am trying to add the same function to scala -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Resolved] (KAFKA-9483) Add Scala KStream#toTable to the Streams DSL
[ https://issues.apache.org/jira/browse/KAFKA-9483?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Matthias J. Sax resolved KAFKA-9483. Fix Version/s: 2.5.0 Resolution: Fixed > Add Scala KStream#toTable to the Streams DSL > > > Key: KAFKA-9483 > URL: https://issues.apache.org/jira/browse/KAFKA-9483 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: highluck >Assignee: highluck >Priority: Minor > Fix For: 2.5.0 > > > [KIP-523: Add KStream#toTable to the Streams > DSL|https://cwiki.apache.org/confluence/display/KAFKA/KIP-523%3A+Add+KStream%23toTable+to+the+Streams+DSL?src=jira] > > I am trying to add the same function to scala -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-9540) Application getting "Could not find the standby task 0_4 while closing it" error
[ https://issues.apache.org/jira/browse/KAFKA-9540?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034911#comment-17034911 ] Sophie Blee-Goldman commented on KAFKA-9540: Hey [~badai], thanks for the ticket. Your analysis is correct that this should not be logged as an error, since the cause is a completely valid situation: a standby task is not created if there are no state stores for it to actually work on. That particular standby task will actually never get created, so whichever thread ends up with this task will always hit this upon closing it. It's probably ok to go all the way down to debug, since warn might still suggest to users that something is wrong. I think this is actually fixed in trunk already due to some significant refactoring of the task management code. But I can quick together a quick PR to bump down the log level on 2.4/2.5 (won't make it into 2.5.0 but may get into 2.4.1) Of course the real fix would be for the assignor to be smart enough not to assign these ghost standbys to begin with. We should be able to fix that up as part of KIP-441 > Application getting "Could not find the standby task 0_4 while closing it" > error > > > Key: KAFKA-9540 > URL: https://issues.apache.org/jira/browse/KAFKA-9540 > Project: Kafka > Issue Type: Bug > Components: streams >Reporter: Badai Aqrandista >Priority: Minor > > Because of this the following line, there is a possibility that some standby > tasks might not be created: > https://github.com/apache/kafka/blob/2.4.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java#L436 > Then causing this line to not adding the task to standby task list: > https://github.com/apache/kafka/blob/2.4.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java#L299 > But this line assumes that all standby tasks are to be created and add it to > the standby list: > https://github.com/apache/kafka/blob/2.4.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java#L168 > This results in user getting this error message on the next > PARTITION_ASSIGNMENT state: > {noformat} > Could not find the standby task 0_4 while closing it > (org.apache.kafka.streams.processor.internals.AssignedStandbyTasks:74) > {noformat} > But the harm caused by this issue is minimal: No standby task for some > partitions. And it is recreated on the next rebalance anyway. So, I suggest > lowering this message to WARN. Or probably check to WARN when standby task > could not be created. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-9455) Consider using TreeMap for in-memory stores of Streams
[ https://issues.apache.org/jira/browse/KAFKA-9455?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034913#comment-17034913 ] highluck commented on KAFKA-9455: - @Guozhang Wang Thanks! you Want to add a new point api? > Consider using TreeMap for in-memory stores of Streams > -- > > Key: KAFKA-9455 > URL: https://issues.apache.org/jira/browse/KAFKA-9455 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: Guozhang Wang >Assignee: highluck >Priority: Major > Labels: newbie++ > > From [~ableegoldman]: It's worth noting that it might be a good idea to > switch to TreeMap for different reasons. Right now the ConcurrentSkipListMap > allows us to safely perform range queries without copying over the entire > keyset, but the performance on point queries seems to scale noticeably worse > with the number of unique keys. Point queries are used by aggregations while > range queries are used by windowed joins, but of course both are available > within the PAPI and for interactive queries so it's hard to say which we > should prefer. Maybe rather than make that tradeoff we should have one > version for efficient range queries (a "JoinWindowStore") and one for > efficient point queries ("AggWindowStore") - or something. I know we've had > similar thoughts for a different RocksDB store layout for Joins (although I > can't find that ticket anywhere..), it seems like the in-memory stores could > benefit from a special "Join" version as well cc/ Guozhang Wang > Here are some random thoughts: > 1. For kafka streams processing logic (i.e. without IQ), it's better to make > all processing logic relying on point queries rather than range queries. > Right now the only processor that use range queries are, as mentioned above, > windowed stream-stream joins. I think we should consider using a different > window implementation for this (and as a result also get rid of the > retainDuplicate flags) to refactor the windowed stream-stream join operation. > 2. With 1), range queries would only be exposed as IQ. Depending on its usage > frequency I think it makes lots of sense to optimize for single-point queries. > Of course, even without step 1) we should still consider using tree-map for > windowed in-memory stores to have a better scaling effect. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-9540) Application getting "Could not find the standby task 0_4 while closing it" error
[ https://issues.apache.org/jira/browse/KAFKA-9540?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034915#comment-17034915 ] ASF GitHub Bot commented on KAFKA-9540: --- ableegoldman commented on pull request #8092: KAFKA-9540: Move "Could not find the standby task while closing it" log to debug level URL: https://github.com/apache/kafka/pull/8092 As described in the ticket, this message is logged at the error level but only indicates that a standby task was not created (as is the case if its subtopology is stateless). Moving this to debug level, and clarifying the implications in the log level. Targeting this PR against 2.4, as the issue is incidentally fixed in trunk as part of the tech debt cleanup. We should also merge this fix to 2.5 but need to wait for the release, since this is obviously not a blocker This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Application getting "Could not find the standby task 0_4 while closing it" > error > > > Key: KAFKA-9540 > URL: https://issues.apache.org/jira/browse/KAFKA-9540 > Project: Kafka > Issue Type: Bug > Components: streams >Reporter: Badai Aqrandista >Priority: Minor > > Because of this the following line, there is a possibility that some standby > tasks might not be created: > https://github.com/apache/kafka/blob/2.4.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java#L436 > Then causing this line to not adding the task to standby task list: > https://github.com/apache/kafka/blob/2.4.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java#L299 > But this line assumes that all standby tasks are to be created and add it to > the standby list: > https://github.com/apache/kafka/blob/2.4.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java#L168 > This results in user getting this error message on the next > PARTITION_ASSIGNMENT state: > {noformat} > Could not find the standby task 0_4 while closing it > (org.apache.kafka.streams.processor.internals.AssignedStandbyTasks:74) > {noformat} > But the harm caused by this issue is minimal: No standby task for some > partitions. And it is recreated on the next rebalance anyway. So, I suggest > lowering this message to WARN. Or probably check to WARN when standby task > could not be created. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-9540) Application getting "Could not find the standby task 0_4 while closing it" error
[ https://issues.apache.org/jira/browse/KAFKA-9540?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034918#comment-17034918 ] ASF GitHub Bot commented on KAFKA-9540: --- guozhangwang commented on pull request #8092: KAFKA-9540: Move "Could not find the standby task while closing it" log to debug level URL: https://github.com/apache/kafka/pull/8092 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Application getting "Could not find the standby task 0_4 while closing it" > error > > > Key: KAFKA-9540 > URL: https://issues.apache.org/jira/browse/KAFKA-9540 > Project: Kafka > Issue Type: Bug > Components: streams >Reporter: Badai Aqrandista >Priority: Minor > > Because of this the following line, there is a possibility that some standby > tasks might not be created: > https://github.com/apache/kafka/blob/2.4.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java#L436 > Then causing this line to not adding the task to standby task list: > https://github.com/apache/kafka/blob/2.4.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java#L299 > But this line assumes that all standby tasks are to be created and add it to > the standby list: > https://github.com/apache/kafka/blob/2.4.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java#L168 > This results in user getting this error message on the next > PARTITION_ASSIGNMENT state: > {noformat} > Could not find the standby task 0_4 while closing it > (org.apache.kafka.streams.processor.internals.AssignedStandbyTasks:74) > {noformat} > But the harm caused by this issue is minimal: No standby task for some > partitions. And it is recreated on the next rebalance anyway. So, I suggest > lowering this message to WARN. Or probably check to WARN when standby task > could not be created. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (KAFKA-9540) Application getting "Could not find the standby task 0_4 while closing it" error
[ https://issues.apache.org/jira/browse/KAFKA-9540?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Sophie Blee-Goldman updated KAFKA-9540: --- Fix Version/s: 2.4.1 > Application getting "Could not find the standby task 0_4 while closing it" > error > > > Key: KAFKA-9540 > URL: https://issues.apache.org/jira/browse/KAFKA-9540 > Project: Kafka > Issue Type: Bug > Components: streams >Reporter: Badai Aqrandista >Priority: Minor > Fix For: 2.4.1 > > > Because of this the following line, there is a possibility that some standby > tasks might not be created: > https://github.com/apache/kafka/blob/2.4.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java#L436 > Then causing this line to not adding the task to standby task list: > https://github.com/apache/kafka/blob/2.4.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java#L299 > But this line assumes that all standby tasks are to be created and add it to > the standby list: > https://github.com/apache/kafka/blob/2.4.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java#L168 > This results in user getting this error message on the next > PARTITION_ASSIGNMENT state: > {noformat} > Could not find the standby task 0_4 while closing it > (org.apache.kafka.streams.processor.internals.AssignedStandbyTasks:74) > {noformat} > But the harm caused by this issue is minimal: No standby task for some > partitions. And it is recreated on the next rebalance anyway. So, I suggest > lowering this message to WARN. Or probably check to WARN when standby task > could not be created. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (KAFKA-9540) Application getting "Could not find the standby task 0_4 while closing it" error
[ https://issues.apache.org/jira/browse/KAFKA-9540?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Sophie Blee-Goldman updated KAFKA-9540: --- Affects Version/s: 2.5.0 2.4.0 > Application getting "Could not find the standby task 0_4 while closing it" > error > > > Key: KAFKA-9540 > URL: https://issues.apache.org/jira/browse/KAFKA-9540 > Project: Kafka > Issue Type: Bug > Components: streams >Affects Versions: 2.4.0, 2.5.0 >Reporter: Badai Aqrandista >Priority: Minor > Fix For: 2.4.1 > > > Because of this the following line, there is a possibility that some standby > tasks might not be created: > https://github.com/apache/kafka/blob/2.4.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java#L436 > Then causing this line to not adding the task to standby task list: > https://github.com/apache/kafka/blob/2.4.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java#L299 > But this line assumes that all standby tasks are to be created and add it to > the standby list: > https://github.com/apache/kafka/blob/2.4.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java#L168 > This results in user getting this error message on the next > PARTITION_ASSIGNMENT state: > {noformat} > Could not find the standby task 0_4 while closing it > (org.apache.kafka.streams.processor.internals.AssignedStandbyTasks:74) > {noformat} > But the harm caused by this issue is minimal: No standby task for some > partitions. And it is recreated on the next rebalance anyway. So, I suggest > lowering this message to WARN. Or probably check to WARN when standby task > could not be created. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-9455) Consider using TreeMap for in-memory stores of Streams
[ https://issues.apache.org/jira/browse/KAFKA-9455?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034923#comment-17034923 ] Guozhang Wang commented on KAFKA-9455: -- No we do not need a new point API, the existing one should be good enough. -- -- Guozhang > Consider using TreeMap for in-memory stores of Streams > -- > > Key: KAFKA-9455 > URL: https://issues.apache.org/jira/browse/KAFKA-9455 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: Guozhang Wang >Assignee: highluck >Priority: Major > Labels: newbie++ > > From [~ableegoldman]: It's worth noting that it might be a good idea to > switch to TreeMap for different reasons. Right now the ConcurrentSkipListMap > allows us to safely perform range queries without copying over the entire > keyset, but the performance on point queries seems to scale noticeably worse > with the number of unique keys. Point queries are used by aggregations while > range queries are used by windowed joins, but of course both are available > within the PAPI and for interactive queries so it's hard to say which we > should prefer. Maybe rather than make that tradeoff we should have one > version for efficient range queries (a "JoinWindowStore") and one for > efficient point queries ("AggWindowStore") - or something. I know we've had > similar thoughts for a different RocksDB store layout for Joins (although I > can't find that ticket anywhere..), it seems like the in-memory stores could > benefit from a special "Join" version as well cc/ Guozhang Wang > Here are some random thoughts: > 1. For kafka streams processing logic (i.e. without IQ), it's better to make > all processing logic relying on point queries rather than range queries. > Right now the only processor that use range queries are, as mentioned above, > windowed stream-stream joins. I think we should consider using a different > window implementation for this (and as a result also get rid of the > retainDuplicate flags) to refactor the windowed stream-stream join operation. > 2. With 1), range queries would only be exposed as IQ. Depending on its usage > frequency I think it makes lots of sense to optimize for single-point queries. > Of course, even without step 1) we should still consider using tree-map for > windowed in-memory stores to have a better scaling effect. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-9455) Consider using TreeMap for in-memory stores of Streams
[ https://issues.apache.org/jira/browse/KAFKA-9455?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034927#comment-17034927 ] highluck commented on KAFKA-9455: - @Guozhang Thanks! it was helpful! > Consider using TreeMap for in-memory stores of Streams > -- > > Key: KAFKA-9455 > URL: https://issues.apache.org/jira/browse/KAFKA-9455 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: Guozhang Wang >Assignee: highluck >Priority: Major > Labels: newbie++ > > From [~ableegoldman]: It's worth noting that it might be a good idea to > switch to TreeMap for different reasons. Right now the ConcurrentSkipListMap > allows us to safely perform range queries without copying over the entire > keyset, but the performance on point queries seems to scale noticeably worse > with the number of unique keys. Point queries are used by aggregations while > range queries are used by windowed joins, but of course both are available > within the PAPI and for interactive queries so it's hard to say which we > should prefer. Maybe rather than make that tradeoff we should have one > version for efficient range queries (a "JoinWindowStore") and one for > efficient point queries ("AggWindowStore") - or something. I know we've had > similar thoughts for a different RocksDB store layout for Joins (although I > can't find that ticket anywhere..), it seems like the in-memory stores could > benefit from a special "Join" version as well cc/ Guozhang Wang > Here are some random thoughts: > 1. For kafka streams processing logic (i.e. without IQ), it's better to make > all processing logic relying on point queries rather than range queries. > Right now the only processor that use range queries are, as mentioned above, > windowed stream-stream joins. I think we should consider using a different > window implementation for this (and as a result also get rid of the > retainDuplicate flags) to refactor the windowed stream-stream join operation. > 2. With 1), range queries would only be exposed as IQ. Depending on its usage > frequency I think it makes lots of sense to optimize for single-point queries. > Of course, even without step 1) we should still consider using tree-map for > windowed in-memory stores to have a better scaling effect. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-9390) Non-key joining of KTable not compatible with confluent avro serdes
[ https://issues.apache.org/jira/browse/KAFKA-9390?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034931#comment-17034931 ] ASF GitHub Bot commented on KAFKA-9390: --- vvcephei commented on pull request #8054: KAFKA-9390: Make serde pseudo-topics unique URL: https://github.com/apache/kafka/pull/8054 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Non-key joining of KTable not compatible with confluent avro serdes > --- > > Key: KAFKA-9390 > URL: https://issues.apache.org/jira/browse/KAFKA-9390 > Project: Kafka > Issue Type: Bug > Components: streams >Affects Versions: 2.4.0 >Reporter: Andy Bryant >Assignee: John Roesler >Priority: Blocker > Fix For: 2.5.0, 2.4.1 > > > I was trying out the new one-to-many KTable joins against some CDC data in > Avro format and kept getting serialisation errors. > > {code:java} > org.apache.kafka.common.errors.SerializationException: Error registering Avro > schema: > {"type":"record","name":"Key","namespace":"dbserver1.inventory.orders","fields":[ > {"name":"order_number","type":"int"} > ],"connect.name":"dbserver1.inventory.orders.Key"} > Caused by: > io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: > Schema being registered is incompatible with an earlier schema; error code: > 409 > > {code} > Both tables have avro keys of different types (one is an order key, the other > a customer key). > This looks like it will cause issues. > [https://github.com/apache/kafka/blob/2.4/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/CombinedKeySchema.java#L57-L60] > They will both attempt to register schemas with the same subject to the > schema registry which will fail a backward compatibility check. > I also noticed in the schema registry there were some subjects that didn't > have the application id prefix. This is probably caused by this... > > [https://github.com/apache/kafka/blob/2.4/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/ForeignJoinSubscriptionSendProcessorSupplier.java#L88] > Where here {{repartitionTopicName}} doesn't have the application prefix. > > > > > -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Resolved] (KAFKA-9390) Non-key joining of KTable not compatible with confluent avro serdes
[ https://issues.apache.org/jira/browse/KAFKA-9390?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] John Roesler resolved KAFKA-9390. - Resolution: Fixed > Non-key joining of KTable not compatible with confluent avro serdes > --- > > Key: KAFKA-9390 > URL: https://issues.apache.org/jira/browse/KAFKA-9390 > Project: Kafka > Issue Type: Bug > Components: streams >Affects Versions: 2.4.0 >Reporter: Andy Bryant >Assignee: John Roesler >Priority: Blocker > Fix For: 2.5.0, 2.4.1 > > > I was trying out the new one-to-many KTable joins against some CDC data in > Avro format and kept getting serialisation errors. > > {code:java} > org.apache.kafka.common.errors.SerializationException: Error registering Avro > schema: > {"type":"record","name":"Key","namespace":"dbserver1.inventory.orders","fields":[ > {"name":"order_number","type":"int"} > ],"connect.name":"dbserver1.inventory.orders.Key"} > Caused by: > io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: > Schema being registered is incompatible with an earlier schema; error code: > 409 > > {code} > Both tables have avro keys of different types (one is an order key, the other > a customer key). > This looks like it will cause issues. > [https://github.com/apache/kafka/blob/2.4/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/CombinedKeySchema.java#L57-L60] > They will both attempt to register schemas with the same subject to the > schema registry which will fail a backward compatibility check. > I also noticed in the schema registry there were some subjects that didn't > have the application id prefix. This is probably caused by this... > > [https://github.com/apache/kafka/blob/2.4/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/ForeignJoinSubscriptionSendProcessorSupplier.java#L88] > Where here {{repartitionTopicName}} doesn't have the application prefix. > > > > > -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-9355) RocksDB statistics are removed from JMX when EOS enabled and empty local state dir
[ https://issues.apache.org/jira/browse/KAFKA-9355?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17034963#comment-17034963 ] ASF GitHub Bot commented on KAFKA-9355: --- guozhangwang commented on pull request #7996: KAFKA-9355: Fix bug that removed RocksDB metrics after failure in EOS URL: https://github.com/apache/kafka/pull/7996 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > RocksDB statistics are removed from JMX when EOS enabled and empty local > state dir > -- > > Key: KAFKA-9355 > URL: https://issues.apache.org/jira/browse/KAFKA-9355 > Project: Kafka > Issue Type: Bug > Components: metrics, streams >Affects Versions: 2.4.0 >Reporter: Stanislav Savulchik >Assignee: Bruno Cadonna >Priority: Major > Attachments: metric-removal.log > > > *Steps to Reproduce* > Set processing.guarantee = exactly_once and remove local state dir in order > to force state restoration from changelog topics that have to be non empty. > *Expected Behavior* > There are registered MBeans like > kafka.streams:type=stream-state-metrics,client-id=-678e4b25-8fc7-4266-85a0-7a5fe52a4060-StreamThread-1,task-id=0_0,rocksdb-state-id= > for persistent RocksDB KeyValueStore-s after streams task state restoration. > *Actual Behavior* > There are no registered MBeans like above after streams task state > restoration. > *Details* > I managed to inject custom MetricsReporter in order to log metricChange and > metricRemoval calls. According to the logs at some point the missing metrics > are removed and never restored later. Here is an excerpt for > number-open-files metric: > {noformat} > 16:33:40.403 DEBUG c.m.r.LoggingMetricsReporter - metricChange MetricName > [name=number-open-files, group=stream-state-metrics, description=Number of > currently open files, > tags={client-id=morpheus.conversion-678e4b25-8fc7-4266-85a0-7a5fe52a4060-StreamThread-1, > task-id=0_0, rocksdb-state-id=buffered-event}] > 16:33:40.403 DEBUG o.a.k.s.s.i.m.RocksDBMetricsRecorder - [RocksDB Metrics > Recorder for buffered-event] Adding metrics recorder of task 0_0 to metrics > recording trigger > 16:33:40.403 DEBUG o.a.k.s.s.i.m.RocksDBMetricsRecorder - [RocksDB Metrics > Recorder for buffered-event] Adding statistics for store buffered-event of > task 0_0 > 16:33:40.610 INFO o.a.k.s.p.i.StoreChangelogReader - stream-thread > [morpheus.conversion-678e4b25-8fc7-4266-85a0-7a5fe52a4060-StreamThread-1] No > checkpoint found for task 0_0 state store buffered-event changelog > morpheus.conversion-buffered-event-changelog-0 with EOS turned on. > Reinitializing the task and restore its state from the beginning. > 16:33:40.610 DEBUG o.a.k.s.s.i.m.RocksDBMetricsRecorder - [RocksDB Metrics > Recorder for buffered-event] Removing statistics for store buffered-event of > task 0_0 > 16:33:40.610 DEBUG o.a.k.s.s.i.m.RocksDBMetricsRecorder - [RocksDB Metrics > Recorder for buffered-event] Removing metrics recorder for store > buffered-event of task 0_0 from metrics recording trigger > 16:33 > 16:33:40.611 DEBUG c.m.r.LoggingMetricsReporter - metricRemoval MetricName > [name=number-open-files, group=stream-state-metrics, description=Number of > currently open files, > tags={client-id=morpheus.conversion-678e4b25-8fc7-4266-85a0-7a5fe52a4060-StreamThread-1, > task-id=0_0, rocksdb-state-id=buffered-event}] > 16:33:40.625 DEBUG o.a.k.s.s.i.m.RocksDBMetricsRecorder - [RocksDB Metrics > Recorder for buffered-event] Adding metrics recorder of task 0_0 to metrics > recording trigger > 16:33:40.625 DEBUG o.a.k.s.s.i.m.RocksDBMetricsRecorder - [RocksDB Metrics > Recorder for buffered-event] Adding statistics for store buffered-event of > task 0_0 > ... > (no more calls to metricChange for the removed number-open-files > metric){noformat} > Also a complete log is attached [^metric-removal.log] > Metric removal happens along this call stack: > {noformat} > 19:27:35.509 DEBUG c.m.r.LoggingMetricsReporter - metricRemoval MetricName > [name=number-open-files, group=stream-state-metrics, description=Number of > currently open files, > tags={client-id=morpheus.conversion-9b76f302-7149-47de-b17b-362d642e05d5-StreamThread-1, > task-id=0_0, rocksdb-state-id=buffered-event}] > java.lang.Exception: null >at > casino.morpheus.reporter.LoggingMetricsReporter.metricRemoval(LoggingMetricsReporter.scala:24) >at org.apache.kafka.common.metrics.Metrics.removeMetric(Metrics.java:534) >at org.apache.kafka.common.me
[jira] [Resolved] (KAFKA-9487) Followup : KAFKA-9445(Allow fetching a key from a single partition); addressing code review comments
[ https://issues.apache.org/jira/browse/KAFKA-9487?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Navinder Brar resolved KAFKA-9487. -- Resolution: Fixed > Followup : KAFKA-9445(Allow fetching a key from a single partition); > addressing code review comments > > > Key: KAFKA-9487 > URL: https://issues.apache.org/jira/browse/KAFKA-9487 > Project: Kafka > Issue Type: New Feature > Components: streams >Reporter: Navinder Brar >Assignee: Navinder Brar >Priority: Blocker > Fix For: 2.5.0 > > > A few code review comments are left to be addressed from Kafka 9445, which I > will be addressing in this PR. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (KAFKA-9541) Flaky Test DescribeConsumerGroupTest#testDescribeGroupWithShortInitializationTimeout
huxihx created KAFKA-9541: - Summary: Flaky Test DescribeConsumerGroupTest#testDescribeGroupWithShortInitializationTimeout Key: KAFKA-9541 URL: https://issues.apache.org/jira/browse/KAFKA-9541 Project: Kafka Issue Type: Bug Components: core, unit tests Affects Versions: 2.4.0 Reporter: huxihx Assignee: huxihx h3. Error Message java.lang.AssertionError: assertion failed h3. Stacktrace java.lang.AssertionError: assertion failed at scala.Predef$.assert(Predef.scala:267) at kafka.admin.DescribeConsumerGroupTest.testDescribeGroupMembersWithShortInitializationTimeout(DescribeConsumerGroupTest.scala:630) 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.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26) at org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27) at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306) at org.junit.runners.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100) at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:366) 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:331) at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79) at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:329) at org.junit.runners.ParentRunner.access$100(ParentRunner.java:66) at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:293) at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26) at org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27) at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306) at org.junit.runners.ParentRunner.run(ParentRunner.java:413) 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 jdk.internal.reflect.GeneratedMethodAccessor13.invoke(Unknown Source) at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.base/java.lang.reflect.Method.invoke(Method.java:566) at org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:36) at org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24) at org.gradle.internal.dispatch.ContextClassLoaderDispatch.dispatch(ContextClassLoaderDispatch.java:33) at org.gradle.internal.dispatch.ProxyDispatchAdapter$DispatchingInvocationHandler.invoke(ProxyDispatchAdapter.java:94) at com.sun.proxy.$Proxy2.processTestClass(Unknown Source) at org.gradle.api.internal.tasks.testing.worker.TestWorker.processTestClass(TestWorker.java:118) at jdk.internal.reflect.GeneratedMethodAccessor12.invoke(Unknown Source) at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.base/java.lang.reflect.Method.invoke(Method.java:566) at org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:36) at org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24) at org.gradle.internal.remote.internal.hub.MessageHubBackedObjectConnection$DispatchWrapper.dispatch(MessageHubBackedObjectConnection.java:182) at org.gradle.internal.remote.internal.hub.MessageHubBackedObjectConnection$DispatchWrapper.dispatch(MessageHubBackedObjectConnection.java:164) at org.gradle.internal.remote.internal.hub.MessageHub$Handler.run(MessageHub.java:412) at org.gradle.internal.concurrent.ExecutorPolicy$CatchAndRecordFailures.onExecute(ExecutorPolicy.java:64) at org.gradle.internal.c
[jira] [Commented] (KAFKA-9541) Flaky Test DescribeConsumerGroupTest#testDescribeGroupWithShortInitializationTimeout
[ https://issues.apache.org/jira/browse/KAFKA-9541?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17035028#comment-17035028 ] huxihx commented on KAFKA-9541: --- Occasionally the captured exception is DisconnectedException instead of TimeoutException. That might be due to an unexpected long pause that caused the node disconnection. > Flaky Test > DescribeConsumerGroupTest#testDescribeGroupWithShortInitializationTimeout > > > Key: KAFKA-9541 > URL: https://issues.apache.org/jira/browse/KAFKA-9541 > Project: Kafka > Issue Type: Bug > Components: core, unit tests >Affects Versions: 2.4.0 >Reporter: huxihx >Assignee: huxihx >Priority: Major > > h3. Error Message > java.lang.AssertionError: assertion failed > h3. Stacktrace > java.lang.AssertionError: assertion failed at > scala.Predef$.assert(Predef.scala:267) at > kafka.admin.DescribeConsumerGroupTest.testDescribeGroupMembersWithShortInitializationTimeout(DescribeConsumerGroupTest.scala:630) > 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.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26) > at > org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27) > at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306) at > org.junit.runners.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100) > at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:366) 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:331) at > org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79) at > org.junit.runners.ParentRunner.runChildren(ParentRunner.java:329) at > org.junit.runners.ParentRunner.access$100(ParentRunner.java:66) at > org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:293) at > org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26) > at > org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27) > at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306) at > org.junit.runners.ParentRunner.run(ParentRunner.java:413) 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 jdk.internal.reflect.GeneratedMethodAccessor13.invoke(Unknown Source) at > java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) > at java.base/java.lang.reflect.Method.invoke(Method.java:566) at > org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:36) > at > org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24) > at > org.gradle.internal.dispatch.ContextClassLoaderDispatch.dispatch(ContextClassLoaderDispatch.java:33) > at > org.gradle.internal.dispatch.ProxyDispatchAdapter$DispatchingInvocationHandler.invoke(ProxyDispatchAdapter.java:94) > at com.sun.proxy.$Proxy2.processTestClass(Unknown Source) at > org.gradle.api.internal.tasks.testing.worker.TestWorker.processTestClass(TestWorker.java:118) > at jdk.internal.reflect.GeneratedMethodAccessor12.invoke(Unknown Source) at > java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) > at java.base/java.lang.reflect.Method.invoke(Method.java:566) at > org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:36) > at > or
[jira] [Updated] (KAFKA-9541) Flaky Test DescribeConsumerGroupTest#testDescribeGroupMembersWithShortInitializationTimeout
[ https://issues.apache.org/jira/browse/KAFKA-9541?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] huxihx updated KAFKA-9541: -- Summary: Flaky Test DescribeConsumerGroupTest#testDescribeGroupMembersWithShortInitializationTimeout (was: Flaky Test DescribeConsumerGroupTest#testDescribeGroupWithShortInitializationTimeout) > Flaky Test > DescribeConsumerGroupTest#testDescribeGroupMembersWithShortInitializationTimeout > --- > > Key: KAFKA-9541 > URL: https://issues.apache.org/jira/browse/KAFKA-9541 > Project: Kafka > Issue Type: Bug > Components: core, unit tests >Affects Versions: 2.4.0 >Reporter: huxihx >Assignee: huxihx >Priority: Major > > h3. Error Message > java.lang.AssertionError: assertion failed > h3. Stacktrace > java.lang.AssertionError: assertion failed at > scala.Predef$.assert(Predef.scala:267) at > kafka.admin.DescribeConsumerGroupTest.testDescribeGroupMembersWithShortInitializationTimeout(DescribeConsumerGroupTest.scala:630) > 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.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26) > at > org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27) > at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306) at > org.junit.runners.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100) > at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:366) 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:331) at > org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79) at > org.junit.runners.ParentRunner.runChildren(ParentRunner.java:329) at > org.junit.runners.ParentRunner.access$100(ParentRunner.java:66) at > org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:293) at > org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26) > at > org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27) > at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306) at > org.junit.runners.ParentRunner.run(ParentRunner.java:413) 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 jdk.internal.reflect.GeneratedMethodAccessor13.invoke(Unknown Source) at > java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) > at java.base/java.lang.reflect.Method.invoke(Method.java:566) at > org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:36) > at > org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24) > at > org.gradle.internal.dispatch.ContextClassLoaderDispatch.dispatch(ContextClassLoaderDispatch.java:33) > at > org.gradle.internal.dispatch.ProxyDispatchAdapter$DispatchingInvocationHandler.invoke(ProxyDispatchAdapter.java:94) > at com.sun.proxy.$Proxy2.processTestClass(Unknown Source) at > org.gradle.api.internal.tasks.testing.worker.TestWorker.processTestClass(TestWorker.java:118) > at jdk.internal.reflect.GeneratedMethodAccessor12.invoke(Unknown Source) at > java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) > at java.base/java.lang.reflect.Method.invoke(Method.java:566) at > org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:36) > at > org.gradle.internal
[jira] [Resolved] (KAFKA-9503) TopologyTestDriver processes intermediate results in the wrong order
[ https://issues.apache.org/jira/browse/KAFKA-9503?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] John Roesler resolved KAFKA-9503. - Resolution: Fixed > TopologyTestDriver processes intermediate results in the wrong order > > > Key: KAFKA-9503 > URL: https://issues.apache.org/jira/browse/KAFKA-9503 > Project: Kafka > Issue Type: Bug > Components: streams-test-utils >Reporter: John Roesler >Assignee: John Roesler >Priority: Major > > TopologyTestDriver has the feature that it processes each input > synchronously, resolving one of the most significant challenges with > verifying the correctness of streaming applications. > When processing an input, it feeds that record to the source node, which then > synchronously (it's always synchronous within a task) gets passed through the > subtopology via Context#forward calls. Ultimately, outputs from that input > are forwarded into the RecordCollector, which converts it to Producer.send > calls. In TopologyTestDriver, this Producer is a special one that actually > just captures the records. > Some output topics from one subtopology are inputs to another subtopology. > For example, repartition topics. Immediately after the synchronous > subtopology process() invocation, TopologyTestDriver iterates over the > collected outputs from the special Producer. If they are purely output > records, it just enqueues them for later retrieval by testing code. If they > are records for internal topics, though, TopologyTestDriver immediately > processes them as inputs for the relevant subtopology. > The problem, and this is very subtle, is that TopologyTestDriver does this > recursively, which with some (apparently rare) programs can cause the output > to be observed in an invalid order. > One such program is the one I wrote to test the fix for KAFKA-9487 . It > involves a foreign-key join whose result is joined back to one of its inputs. > {noformat} > Here's a simplified version: > // foreign key join > J = A.join(B, (extractor) a -> a.b, (joiner) (a,b) -> new Pair(a, b)) > // equi-join > OUT = A.join(J, (joiner) (a, j) -> new Pair(a, j)) > Let's say we have the following initial condition: > A: > a1 = {v: X, b: b1} > B: > b1 = {v: Y} > J: > a1 = Pair({v: X}, b: b1}, {v: Y}) > OUT: > a1 = Pair({v: X}, b: b1}, Pair({v: X}, b: b1}, {v: Y})) > Now, piping an update: > a1: {v: Z, b: b1} > results immediately in two buffered results in the Producer: > (FK join subscription): b1: {a1} > (OUT): a1 = Pair({v: Z}, b: b1}, Pair({v: X}, b: b1}, {v: Y})) > Note that the FK join result isn't updated synchronously, since it's an async > operation, so the RHS lookup is temporarily incorrect, yielding the nonsense > intermediate result where the outer pair has the updated value for a1, but > the inner (fk result) one still has the old value for a1. > However! We don't buffer that output record for consumption by testing code > yet, we leave it in the internal Producer while we process the first > intermediate record (the FK subscription). > Processing that internal record means that we have a new internal record to > process: > (FK join subscription response): a1: {b1: {v: Y}} > so right now, our internal-records-to-process stack looks like: > (FK join subscription response): a1: {b1: {v: Y}} > (OUT) a1: Pair({v: Z}, b: b1}, Pair({v: X}, b: b1}, {v: Y})) > Again, we start by processing the first thing, the FK join response, which > results in an updated FK join result: > (J) a1: Pair({v: Z}, b: b1}, {v: Y}) > and output: > (OUT) a1: Pair({v: Z}, b: b1}, Pair({v: Z}, b: b1}, {v: Y})) > and, we still haven't handled the earlier output, so now our > internal-records-to-process stack looks like: > (J) a1: Pair({v: Z}, b: b1}, {v: Y}) > (OUT) a1: Pair({v: Z}, b: b1}, Pair({v: Z}, b: b1}, {v: Y})) > (OUT) a1: Pair({v: Z}, b: b1}, Pair({v: X}, b: b1}, {v: Y})) > At this point, there's nothing else to process in internal topics, so we just > copy the records one by one to the "output" collection for later handling by > testing code, but this yields the wrong final state of: > (OUT) a1: Pair({v: Z}, b: b1}, Pair({v: X}, b: b1}, {v: Y})) > That was an incorrect intermediate result, but because we're processing > internal records recursively (as a stack), it winds up emitted at the end > instead of in the middle. > If we change the processing model from a stack to a queue, the correct order > is preserved, and the final state is: > (OUT) a1: Pair({v: Z}, b: b1}, Pair({v: Z}, b: b1}, {v: Y})) > {noformat} > This is what I did in https://github.com/apache/kafka/pull/8015 -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-9541) Flaky Test DescribeConsumerGroupTest#testDescribeGroupMembersWithShortInitializationTimeout
[ https://issues.apache.org/jira/browse/KAFKA-9541?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17035042#comment-17035042 ] ASF GitHub Bot commented on KAFKA-9541: --- huxihx commented on pull request #8094: KAFKA-9541:Flaky Test DescribeConsumerGroupTest#testDescribeGroupMembersWithShortInitializationTimeout URL: https://github.com/apache/kafka/pull/8094 https://issues.apache.org/jira/browse/KAFKA-9541 Occasionally the captured exception is DisconnectedException instead of TimeoutException. That might be due to an unexpected long pause that caused the node disconnection. *More detailed description of your change, if necessary. The PR title and PR message become the squashed commit message, so use a separate comment to ping reviewers.* *Summary of testing strategy (including rationale) for the feature or bug fix. Unit and/or integration tests are expected for any behaviour change and system tests should be considered for larger changes.* ### Committer Checklist (excluded from commit message) - [ ] Verify design and implementation - [ ] Verify test coverage and CI build status - [ ] Verify documentation (including upgrade notes) This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Flaky Test > DescribeConsumerGroupTest#testDescribeGroupMembersWithShortInitializationTimeout > --- > > Key: KAFKA-9541 > URL: https://issues.apache.org/jira/browse/KAFKA-9541 > Project: Kafka > Issue Type: Bug > Components: core, unit tests >Affects Versions: 2.4.0 >Reporter: huxihx >Assignee: huxihx >Priority: Major > > h3. Error Message > java.lang.AssertionError: assertion failed > h3. Stacktrace > java.lang.AssertionError: assertion failed at > scala.Predef$.assert(Predef.scala:267) at > kafka.admin.DescribeConsumerGroupTest.testDescribeGroupMembersWithShortInitializationTimeout(DescribeConsumerGroupTest.scala:630) > 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.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26) > at > org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27) > at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306) at > org.junit.runners.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100) > at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:366) 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:331) at > org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79) at > org.junit.runners.ParentRunner.runChildren(ParentRunner.java:329) at > org.junit.runners.ParentRunner.access$100(ParentRunner.java:66) at > org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:293) at > org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26) > at > org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27) > at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306) at > org.junit.runners.ParentRunner.run(ParentRunner.java:413) 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.processTestC
[jira] [Commented] (KAFKA-9500) Foreign-Key Join creates an invalid topology
[ https://issues.apache.org/jira/browse/KAFKA-9500?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17035043#comment-17035043 ] ASF GitHub Bot commented on KAFKA-9500: --- vvcephei commented on pull request #8015: KAFKA-9500: Fix topology bug in foreign key joins URL: https://github.com/apache/kafka/pull/8015 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Foreign-Key Join creates an invalid topology > > > Key: KAFKA-9500 > URL: https://issues.apache.org/jira/browse/KAFKA-9500 > Project: Kafka > Issue Type: Bug > Components: streams >Affects Versions: 2.4.0 >Reporter: John Roesler >Assignee: John Roesler >Priority: Blocker > Fix For: 2.5.0, 2.4.1 > > > Foreign-Key Join results are not required to be materialized by default, but > they might be needed if downstream operators need to perform lookups on the > result (such as when the join result participates in an equi-join). > Currently, if the result is explicitly materialized (via Materialized), this > works correctly, but if the result is _not_ materialized explicitly, but _is_ > needed, the topology builder throws an exception that the result store isn't > added to the topology. This was an oversight in testing and review and needs > to be fixed ASAP. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Resolved] (KAFKA-9500) Foreign-Key Join creates an invalid topology
[ https://issues.apache.org/jira/browse/KAFKA-9500?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] John Roesler resolved KAFKA-9500. - Resolution: Fixed > Foreign-Key Join creates an invalid topology > > > Key: KAFKA-9500 > URL: https://issues.apache.org/jira/browse/KAFKA-9500 > Project: Kafka > Issue Type: Bug > Components: streams >Affects Versions: 2.4.0 >Reporter: John Roesler >Assignee: John Roesler >Priority: Blocker > Fix For: 2.5.0, 2.4.1 > > > Foreign-Key Join results are not required to be materialized by default, but > they might be needed if downstream operators need to perform lookups on the > result (such as when the join result participates in an equi-join). > Currently, if the result is explicitly materialized (via Materialized), this > works correctly, but if the result is _not_ materialized explicitly, but _is_ > needed, the topology builder throws an exception that the result store isn't > added to the topology. This was an oversight in testing and review and needs > to be fixed ASAP. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (KAFKA-9503) TopologyTestDriver processes intermediate results in the wrong order
[ https://issues.apache.org/jira/browse/KAFKA-9503?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Matthias J. Sax updated KAFKA-9503: --- Fix Version/s: 2.4.1 2.5.0 > TopologyTestDriver processes intermediate results in the wrong order > > > Key: KAFKA-9503 > URL: https://issues.apache.org/jira/browse/KAFKA-9503 > Project: Kafka > Issue Type: Bug > Components: streams-test-utils >Reporter: John Roesler >Assignee: John Roesler >Priority: Major > Fix For: 2.5.0, 2.4.1 > > > TopologyTestDriver has the feature that it processes each input > synchronously, resolving one of the most significant challenges with > verifying the correctness of streaming applications. > When processing an input, it feeds that record to the source node, which then > synchronously (it's always synchronous within a task) gets passed through the > subtopology via Context#forward calls. Ultimately, outputs from that input > are forwarded into the RecordCollector, which converts it to Producer.send > calls. In TopologyTestDriver, this Producer is a special one that actually > just captures the records. > Some output topics from one subtopology are inputs to another subtopology. > For example, repartition topics. Immediately after the synchronous > subtopology process() invocation, TopologyTestDriver iterates over the > collected outputs from the special Producer. If they are purely output > records, it just enqueues them for later retrieval by testing code. If they > are records for internal topics, though, TopologyTestDriver immediately > processes them as inputs for the relevant subtopology. > The problem, and this is very subtle, is that TopologyTestDriver does this > recursively, which with some (apparently rare) programs can cause the output > to be observed in an invalid order. > One such program is the one I wrote to test the fix for KAFKA-9487 . It > involves a foreign-key join whose result is joined back to one of its inputs. > {noformat} > Here's a simplified version: > // foreign key join > J = A.join(B, (extractor) a -> a.b, (joiner) (a,b) -> new Pair(a, b)) > // equi-join > OUT = A.join(J, (joiner) (a, j) -> new Pair(a, j)) > Let's say we have the following initial condition: > A: > a1 = {v: X, b: b1} > B: > b1 = {v: Y} > J: > a1 = Pair({v: X}, b: b1}, {v: Y}) > OUT: > a1 = Pair({v: X}, b: b1}, Pair({v: X}, b: b1}, {v: Y})) > Now, piping an update: > a1: {v: Z, b: b1} > results immediately in two buffered results in the Producer: > (FK join subscription): b1: {a1} > (OUT): a1 = Pair({v: Z}, b: b1}, Pair({v: X}, b: b1}, {v: Y})) > Note that the FK join result isn't updated synchronously, since it's an async > operation, so the RHS lookup is temporarily incorrect, yielding the nonsense > intermediate result where the outer pair has the updated value for a1, but > the inner (fk result) one still has the old value for a1. > However! We don't buffer that output record for consumption by testing code > yet, we leave it in the internal Producer while we process the first > intermediate record (the FK subscription). > Processing that internal record means that we have a new internal record to > process: > (FK join subscription response): a1: {b1: {v: Y}} > so right now, our internal-records-to-process stack looks like: > (FK join subscription response): a1: {b1: {v: Y}} > (OUT) a1: Pair({v: Z}, b: b1}, Pair({v: X}, b: b1}, {v: Y})) > Again, we start by processing the first thing, the FK join response, which > results in an updated FK join result: > (J) a1: Pair({v: Z}, b: b1}, {v: Y}) > and output: > (OUT) a1: Pair({v: Z}, b: b1}, Pair({v: Z}, b: b1}, {v: Y})) > and, we still haven't handled the earlier output, so now our > internal-records-to-process stack looks like: > (J) a1: Pair({v: Z}, b: b1}, {v: Y}) > (OUT) a1: Pair({v: Z}, b: b1}, Pair({v: Z}, b: b1}, {v: Y})) > (OUT) a1: Pair({v: Z}, b: b1}, Pair({v: X}, b: b1}, {v: Y})) > At this point, there's nothing else to process in internal topics, so we just > copy the records one by one to the "output" collection for later handling by > testing code, but this yields the wrong final state of: > (OUT) a1: Pair({v: Z}, b: b1}, Pair({v: X}, b: b1}, {v: Y})) > That was an incorrect intermediate result, but because we're processing > internal records recursively (as a stack), it winds up emitted at the end > instead of in the middle. > If we change the processing model from a stack to a queue, the correct order > is preserved, and the final state is: > (OUT) a1: Pair({v: Z}, b: b1}, Pair({v: Z}, b: b1}, {v: Y})) > {noformat} > This is what I did in https://github.com/apache/kafka/pull/8015 -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (KAFKA-9509) Fix flaky test MirrorConnectorsIntegrationTest.testReplication
[ https://issues.apache.org/jira/browse/KAFKA-9509?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Matthias J. Sax updated KAFKA-9509: --- Affects Version/s: (was: 2.4.1, 2.5.0) > Fix flaky test MirrorConnectorsIntegrationTest.testReplication > -- > > Key: KAFKA-9509 > URL: https://issues.apache.org/jira/browse/KAFKA-9509 > Project: Kafka > Issue Type: Test > Components: mirrormaker >Affects Versions: 2.4.0 >Reporter: Sanjana Kaundinya >Assignee: Sanjana Kaundinya >Priority: Major > > The test > org.apache.kafka.connect.mirror.MirrorConnectorsIntegrationTest.testReplication > is a flaky test for MirrorMaker 2.0. Its flakiness lies in the timing of > when the connectors and tasks are started up. The fix for this would make it > such that when the connectors are started up, to wait until the REST endpoint > returns a positive number of tasks to be confident that we can start testing. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (KAFKA-9509) Fix flaky test MirrorConnectorsIntegrationTest.testReplication
[ https://issues.apache.org/jira/browse/KAFKA-9509?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Matthias J. Sax updated KAFKA-9509: --- Fix Version/s: 2.5.0 > Fix flaky test MirrorConnectorsIntegrationTest.testReplication > -- > > Key: KAFKA-9509 > URL: https://issues.apache.org/jira/browse/KAFKA-9509 > Project: Kafka > Issue Type: Test > Components: mirrormaker >Affects Versions: 2.4.0 >Reporter: Sanjana Kaundinya >Assignee: Sanjana Kaundinya >Priority: Major > Fix For: 2.5.0 > > > The test > org.apache.kafka.connect.mirror.MirrorConnectorsIntegrationTest.testReplication > is a flaky test for MirrorMaker 2.0. Its flakiness lies in the timing of > when the connectors and tasks are started up. The fix for this would make it > such that when the connectors are started up, to wait until the REST endpoint > returns a positive number of tasks to be confident that we can start testing. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-9479) Describe consumer group --all-groups shows header for each entry
[ https://issues.apache.org/jira/browse/KAFKA-9479?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17035111#comment-17035111 ] Vetle Leinonen-Roeim commented on KAFKA-9479: - Done! [https://github.com/apache/kafka/pull/8096] > Describe consumer group --all-groups shows header for each entry > > > Key: KAFKA-9479 > URL: https://issues.apache.org/jira/browse/KAFKA-9479 > Project: Kafka > Issue Type: Improvement >Reporter: Jason Gustafson >Assignee: Jeff Kim >Priority: Major > Labels: newbie > > When using `bin/kafka-consumer-groups.sh --describe --state --all-groups`, we > print output like the following: > {code} > GROUP COORDINATOR (ID) > ASSIGNMENT-STRATEGY STATE #MEMBERS > group1 localhost:9092 (3) rangeStable 1 > > > GROUP COORDINATOR (ID) > ASSIGNMENT-STRATEGY STATE #MEMBERS > group2 localhost:9092 (3) rangeStable 1 > > > {code} > It would be nice if we did not show the header for every entry. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-9479) Describe consumer group --all-groups shows header for each entry
[ https://issues.apache.org/jira/browse/KAFKA-9479?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17035110#comment-17035110 ] ASF GitHub Bot commented on KAFKA-9479: --- vetler commented on pull request #8096: KAFKA-9479 Describe consumer groups with --all-groups now prints head… URL: https://github.com/apache/kafka/pull/8096 KAFKA-9479 Describe consumer groups with --all-groups now prints header only once Describing the consumer groups with --all-groups will now only print the header once, instead of once per entry. The maximum coordinator length is computed first, by looping through the GroupStates in advance, to ensure correct cormatting. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Describe consumer group --all-groups shows header for each entry > > > Key: KAFKA-9479 > URL: https://issues.apache.org/jira/browse/KAFKA-9479 > Project: Kafka > Issue Type: Improvement >Reporter: Jason Gustafson >Assignee: Jeff Kim >Priority: Major > Labels: newbie > > When using `bin/kafka-consumer-groups.sh --describe --state --all-groups`, we > print output like the following: > {code} > GROUP COORDINATOR (ID) > ASSIGNMENT-STRATEGY STATE #MEMBERS > group1 localhost:9092 (3) rangeStable 1 > > > GROUP COORDINATOR (ID) > ASSIGNMENT-STRATEGY STATE #MEMBERS > group2 localhost:9092 (3) rangeStable 1 > > > {code} > It would be nice if we did not show the header for every entry. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-9455) Consider using TreeMap for in-memory stores of Streams
[ https://issues.apache.org/jira/browse/KAFKA-9455?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17035117#comment-17035117 ] highluck commented on KAFKA-9455: - [~guozhang] I have one more question What do you think about splitting a WindowStore into two stores? > Consider using TreeMap for in-memory stores of Streams > -- > > Key: KAFKA-9455 > URL: https://issues.apache.org/jira/browse/KAFKA-9455 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: Guozhang Wang >Assignee: highluck >Priority: Major > Labels: newbie++ > > From [~ableegoldman]: It's worth noting that it might be a good idea to > switch to TreeMap for different reasons. Right now the ConcurrentSkipListMap > allows us to safely perform range queries without copying over the entire > keyset, but the performance on point queries seems to scale noticeably worse > with the number of unique keys. Point queries are used by aggregations while > range queries are used by windowed joins, but of course both are available > within the PAPI and for interactive queries so it's hard to say which we > should prefer. Maybe rather than make that tradeoff we should have one > version for efficient range queries (a "JoinWindowStore") and one for > efficient point queries ("AggWindowStore") - or something. I know we've had > similar thoughts for a different RocksDB store layout for Joins (although I > can't find that ticket anywhere..), it seems like the in-memory stores could > benefit from a special "Join" version as well cc/ Guozhang Wang > Here are some random thoughts: > 1. For kafka streams processing logic (i.e. without IQ), it's better to make > all processing logic relying on point queries rather than range queries. > Right now the only processor that use range queries are, as mentioned above, > windowed stream-stream joins. I think we should consider using a different > window implementation for this (and as a result also get rid of the > retainDuplicate flags) to refactor the windowed stream-stream join operation. > 2. With 1), range queries would only be exposed as IQ. Depending on its usage > frequency I think it makes lots of sense to optimize for single-point queries. > Of course, even without step 1) we should still consider using tree-map for > windowed in-memory stores to have a better scaling effect. -- This message was sent by Atlassian Jira (v8.3.4#803005)