[jira] [Commented] (KAFKA-9533) ValueTransform forwards `null` values

[Bruno Cadonna (Jira)]

[ 
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

[ASF GitHub Bot (Jira)]

[ 
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

[highluck (Jira)]

 [ 
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

[highluck (Jira)]

[ 
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

[Ron Dagostino (Jira)]

[ 
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

[Ismael Juma (Jira)]

[ 
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

[Stanislav Kozlovski (Jira)]

[ 
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

[ASF GitHub Bot (Jira)]

[ 
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

[Jun Rao (Jira)]

[ 
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

[John Roesler (Jira)]

[ 
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

[Matthias J. Sax (Jira)]

[ 
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 (Jira)]

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

[ASF GitHub Bot (Jira)]

[ 
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

[ASF GitHub Bot (Jira)]

[ 
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.

[ASF GitHub Bot (Jira)]

[ 
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

[Sophie Blee-Goldman (Jira)]

[ 
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

[Boyang Chen (Jira)]

 [ 
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

[Boyang Chen (Jira)]

 [ 
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

[Boyang Chen (Jira)]

 [ 
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 (Jira)]

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

[Jason Gustafson (Jira)]

 [ 
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`

[Jason Gustafson (Jira)]

 [ 
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

[Sophie Blee-Goldman (Jira)]

[ 
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

[ASF GitHub Bot (Jira)]

[ 
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

[Boyang Chen (Jira)]

 [ 
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

[Randall Hauch (Jira)]

 [ 
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 (Jira)]

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

[Jason Gustafson (Jira)]

[ 
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

[Ryanne Dolan (Jira)]

[ 
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

[Jeff Kim (Jira)]

[ 
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

[ASF GitHub Bot (Jira)]

[ 
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

[ASF GitHub Bot (Jira)]

[ 
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

[Matthias J. Sax (Jira)]

[ 
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

[Matthias J. Sax (Jira)]

[ 
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

[Matthias J. Sax (Jira)]

 [ 
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

[Matthias J. Sax (Jira)]

[ 
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

[Guozhang Wang (Jira)]

[ 
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 (Jira)]

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

[ASF GitHub Bot (Jira)]

[ 
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

[Matthias J. Sax (Jira)]

 [ 
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

[Sophie Blee-Goldman (Jira)]

[ 
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

[highluck (Jira)]

[ 
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

[ASF GitHub Bot (Jira)]

[ 
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

[ASF GitHub Bot (Jira)]

[ 
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

[Sophie Blee-Goldman (Jira)]

 [ 
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

[Sophie Blee-Goldman (Jira)]

 [ 
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

[Guozhang Wang (Jira)]

[ 
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

[highluck (Jira)]

[ 
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

[ASF GitHub Bot (Jira)]

[ 
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

[John Roesler (Jira)]

 [ 
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

[ASF GitHub Bot (Jira)]

[ 
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

[Navinder Brar (Jira)]

 [ 
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 (Jira)]

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

[huxihx (Jira)]

[ 
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

[huxihx (Jira)]

 [ 
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

[John Roesler (Jira)]

 [ 
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

[ASF GitHub Bot (Jira)]

[ 
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

[ASF GitHub Bot (Jira)]

[ 
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

[John Roesler (Jira)]

 [ 
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

[Matthias J. Sax (Jira)]

 [ 
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

[Matthias J. Sax (Jira)]

 [ 
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

[Matthias J. Sax (Jira)]

 [ 
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

[Vetle Leinonen-Roeim (Jira)]

[ 
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

[ASF GitHub Bot (Jira)]

[ 
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

[highluck (Jira)]

[ 
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)