[jira] [Commented] (KAFKA-7131) Update release script to generate announcement email text
[ https://issues.apache.org/jira/browse/KAFKA-7131?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16624496#comment-16624496 ] bibin sebastian commented on KAFKA-7131: [~mjsax], [~ewencp] can you please take a look? > Update release script to generate announcement email text > - > > Key: KAFKA-7131 > URL: https://issues.apache.org/jira/browse/KAFKA-7131 > Project: Kafka > Issue Type: Improvement >Reporter: Matthias J. Sax >Assignee: bibin sebastian >Priority: Minor > Labels: newbie > > When a release is finalized, we send out an email to announce the release. > Atm, we have a template in the wiki > ([https://cwiki.apache.org/confluence/display/KAFKA/Release+Process|https://cwiki.apache.org/confluence/display/KAFKA/Release+Process]). > However, the template needs some manual changes to fill in the release > number, number of contributors, etc. > Some parts could be automated – the corresponding commands are document in > the wiki already. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-7216) Exception while running kafka-acls.sh from 1.0 env on target Kafka env with 1.1.1
[
https://issues.apache.org/jira/browse/KAFKA-7216?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16624457#comment-16624457
]
ASF GitHub Bot commented on KAFKA-7216:
---
omkreddy opened a new pull request #5680: KAFKA-7216: Ignore unknown
ResourceTypes while loading acl cache
URL: https://github.com/apache/kafka/pull/5680
### 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 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
> Exception while running kafka-acls.sh from 1.0 env on target Kafka env with
> 1.1.1
> -
>
> Key: KAFKA-7216
> URL: https://issues.apache.org/jira/browse/KAFKA-7216
> Project: Kafka
> Issue Type: Bug
> Components: admin
>Affects Versions: 1.1.0, 1.1.1, 2.0.0
>Reporter: Satish Duggana
>Assignee: Manikumar
>Priority: Major
>
> When `kafka-acls.sh` with SimpleAclAuthorizer on target Kafka cluster with
> 1.1.1 version, it throws the below error.
> {code:java}
> kafka.common.KafkaException: DelegationToken not a valid resourceType name.
> The valid names are Topic,Group,Cluster,TransactionalId
> at
> kafka.security.auth.ResourceType$$anonfun$fromString$1.apply(ResourceType.scala:56)
> at
> kafka.security.auth.ResourceType$$anonfun$fromString$1.apply(ResourceType.scala:56)
> at scala.Option.getOrElse(Option.scala:121)
> at kafka.security.auth.ResourceType$.fromString(ResourceType.scala:56)
> at
> kafka.security.auth.SimpleAclAuthorizer$$anonfun$loadCache$1$$anonfun$apply$mcV$sp$1.apply(SimpleAclAuthorizer.scala:233)
> at
> kafka.security.auth.SimpleAclAuthorizer$$anonfun$loadCache$1$$anonfun$apply$mcV$sp$1.apply(SimpleAclAuthorizer.scala:232)
> at scala.collection.Iterator$class.foreach(Iterator.scala:891)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
> at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
> at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
> at
> kafka.security.auth.SimpleAclAuthorizer$$anonfun$loadCache$1.apply$mcV$sp(SimpleAclAuthorizer.scala:232)
> at
> kafka.security.auth.SimpleAclAuthorizer$$anonfun$loadCache$1.apply(SimpleAclAuthorizer.scala:230)
> at
> kafka.security.auth.SimpleAclAuthorizer$$anonfun$loadCache$1.apply(SimpleAclAuthorizer.scala:230)
> at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:216)
> at kafka.utils.CoreUtils$.inWriteLock(CoreUtils.scala:224)
> at
> kafka.security.auth.SimpleAclAuthorizer.loadCache(SimpleAclAuthorizer.scala:230)
> at
> kafka.security.auth.SimpleAclAuthorizer.configure(SimpleAclAuthorizer.scala:114)
> at kafka.admin.AclCommand$.withAuthorizer(AclCommand.scala:83)
> at kafka.admin.AclCommand$.addAcl(AclCommand.scala:93)
> at kafka.admin.AclCommand$.main(AclCommand.scala:53)
> at kafka.admin.AclCommand.main(AclCommand.scala)
> {code}
>
> This is because it tries to get all the resource types registered from ZK
> path and it throws error when `DelegationToken` resource is not defined in
> `ResourceType` of client's Kafka version(which is earlier than 1.1.x)
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-7216) Exception while running kafka-acls.sh from 1.0 env on target Kafka env with 1.1.1
[
https://issues.apache.org/jira/browse/KAFKA-7216?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16624448#comment-16624448
]
ASF GitHub Bot commented on KAFKA-7216:
---
omkreddy opened a new pull request #5679: KAFKA-7216: Ignore unknown
ResourceTypes while loading acl cache
URL: https://github.com/apache/kafka/pull/5679
### 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 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
> Exception while running kafka-acls.sh from 1.0 env on target Kafka env with
> 1.1.1
> -
>
> Key: KAFKA-7216
> URL: https://issues.apache.org/jira/browse/KAFKA-7216
> Project: Kafka
> Issue Type: Bug
> Components: admin
>Affects Versions: 1.1.0, 1.1.1, 2.0.0
>Reporter: Satish Duggana
>Assignee: Manikumar
>Priority: Major
>
> When `kafka-acls.sh` with SimpleAclAuthorizer on target Kafka cluster with
> 1.1.1 version, it throws the below error.
> {code:java}
> kafka.common.KafkaException: DelegationToken not a valid resourceType name.
> The valid names are Topic,Group,Cluster,TransactionalId
> at
> kafka.security.auth.ResourceType$$anonfun$fromString$1.apply(ResourceType.scala:56)
> at
> kafka.security.auth.ResourceType$$anonfun$fromString$1.apply(ResourceType.scala:56)
> at scala.Option.getOrElse(Option.scala:121)
> at kafka.security.auth.ResourceType$.fromString(ResourceType.scala:56)
> at
> kafka.security.auth.SimpleAclAuthorizer$$anonfun$loadCache$1$$anonfun$apply$mcV$sp$1.apply(SimpleAclAuthorizer.scala:233)
> at
> kafka.security.auth.SimpleAclAuthorizer$$anonfun$loadCache$1$$anonfun$apply$mcV$sp$1.apply(SimpleAclAuthorizer.scala:232)
> at scala.collection.Iterator$class.foreach(Iterator.scala:891)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
> at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
> at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
> at
> kafka.security.auth.SimpleAclAuthorizer$$anonfun$loadCache$1.apply$mcV$sp(SimpleAclAuthorizer.scala:232)
> at
> kafka.security.auth.SimpleAclAuthorizer$$anonfun$loadCache$1.apply(SimpleAclAuthorizer.scala:230)
> at
> kafka.security.auth.SimpleAclAuthorizer$$anonfun$loadCache$1.apply(SimpleAclAuthorizer.scala:230)
> at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:216)
> at kafka.utils.CoreUtils$.inWriteLock(CoreUtils.scala:224)
> at
> kafka.security.auth.SimpleAclAuthorizer.loadCache(SimpleAclAuthorizer.scala:230)
> at
> kafka.security.auth.SimpleAclAuthorizer.configure(SimpleAclAuthorizer.scala:114)
> at kafka.admin.AclCommand$.withAuthorizer(AclCommand.scala:83)
> at kafka.admin.AclCommand$.addAcl(AclCommand.scala:93)
> at kafka.admin.AclCommand$.main(AclCommand.scala:53)
> at kafka.admin.AclCommand.main(AclCommand.scala)
> {code}
>
> This is because it tries to get all the resource types registered from ZK
> path and it throws error when `DelegationToken` resource is not defined in
> `ResourceType` of client's Kafka version(which is earlier than 1.1.x)
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-7415) OffsetsForLeaderEpoch may incorrectly respond with undefined epoch causing truncation to HW
[
https://issues.apache.org/jira/browse/KAFKA-7415?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16624312#comment-16624312
]
ASF GitHub Bot commented on KAFKA-7415:
---
hachikuji opened a new pull request #5678: KAFKA-7415; Persist leader epoch and
start offset on becoming a leader
URL: https://github.com/apache/kafka/pull/5678
This patch ensures that the leader epoch cache is updated when a broker
becomes leader with the latest epoch and the log end offset as its starting
offset. This guarantees that the leader will be able to provide the right
truncation point even if the follower has data from leader epochs which the
leader itself does not have. This situation can occur when there are back to
back leader elections.
Additionally, we have made the following changes:
1. The leader epoch cache enforces monotonically increase epochs and
starting offsets among its entry. Whenever a new entry is appended which
violates requirement, we remove the conflicting entries from the cache.
2. Previously we returned an unknown epoch and offset if an epoch is queried
which comes before the first entry in the cache. Now we return the smallest .
For example, if the earliest entry in the cache is (epoch=5, startOffset=10),
then a query for epoch 4 will return (epoch=4, endOffset=10). This ensures that
followers (and consumers in KIP-320) can always determine where the correct
starting point is for the active log range on the leader.
### 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 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
> OffsetsForLeaderEpoch may incorrectly respond with undefined epoch causing
> truncation to HW
> ---
>
> Key: KAFKA-7415
> URL: https://issues.apache.org/jira/browse/KAFKA-7415
> Project: Kafka
> Issue Type: Bug
> Components: replication
>Affects Versions: 2.0.0
>Reporter: Anna Povzner
>Assignee: Jason Gustafson
>Priority: Major
>
> If the follower's last appended epoch is ahead of the leader's last appended
> epoch, the OffsetsForLeaderEpoch response will incorrectly send
> (UNDEFINED_EPOCH, UNDEFINED_EPOCH_OFFSET), and the follower will truncate to
> HW. This may lead to data loss in some rare cases where 2 back-to-back leader
> elections happen (failure of one leader, followed by quick re-election of the
> next leader due to preferred leader election, so that all replicas are still
> in the ISR, and then failure of the 3rd leader).
> The bug is in LeaderEpochFileCache.endOffsetFor(), which returns
> (UNDEFINED_EPOCH, UNDEFINED_EPOCH_OFFSET) if the requested leader epoch is
> ahead of the last leader epoch in the cache. The method should return (last
> leader epoch in the cache, LEO) in this scenario.
> We don't create an entry in a leader epoch cache until a message is appended
> with the new leader epoch. Every append to log calls
> LeaderEpochFileCache.assign(). However, it would be much cleaner if
> `makeLeader` created an entry in the cache as soon as replica becomes a
> leader, which will fix the bug. In case the leader never appends any
> messages, and the next leader epoch starts with the same offset, we already
> have clearAndFlushLatest() that clears entries with start offsets greater or
> equal to the passed offset. LeaderEpochFileCache.assign() could be merged
> with clearAndFlushLatest(), so that we clear cache entries with offsets equal
> or greater than the start offset of the new epoch, so that we do not need to
> call these methods separately.
>
> Here is an example of a scenario where the issue leads to the data loss.
> Suppose we have three replicas: r1, r2, and r3. Initially, the ISR consists
> of (r1, r2, r3) and the leader is r1. The data up to offset 10 has been
> committed to the ISR. Here is the initial state:
> {code:java}
> Leader: r1
> leader epoch: 0
> ISR(r1, r2, r3)
> r1: [hw=10, leo=10]
> r2: [hw=8, leo=10]
> r3: [hw=5, leo=10]
> {code}
> Replica 1 fails and leaves the ISR, which makes Replica 2 the new leader with
> leader epoch = 1. The leader appends a batch, but it is not replicated yet to
> the followers.
> {code:java}
> Leader: r2
> leader epoch: 1
> ISR(r2, r3)
> r1: [hw=10, leo=10]
> r2: [hw=8, leo=11]
> r3: [hw=5, leo=10]
> {code}
> Replica 3 is elected a leader (due to preferred leader election) before it
[jira] [Commented] (KAFKA-7421) Deadlock in Kafka Connect
[
https://issues.apache.org/jira/browse/KAFKA-7421?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16624165#comment-16624165
]
Konstantine Karantasis commented on KAFKA-7421:
---
Thanks [~maver1ck] for the feedback. It makes sense. I have an idea for a fix
and I'll put out a PR soon.
Can you share some info about how you are able to reproduce the issue? Do you
just start the two aforementioned connectors at the same time? I see it's in
distributed mode and that they both get stuck during their startup. Anything
special about their configs?
> Deadlock in Kafka Connect
> -
>
> Key: KAFKA-7421
> URL: https://issues.apache.org/jira/browse/KAFKA-7421
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
>Affects Versions: 2.0.0
>Reporter: Maciej Bryński
>Assignee: Konstantine Karantasis
>Priority: Critical
>
> I'm getting this deadlock on half of Kafka Connect runs when having two
> different types connectors (in this configuration it's debezium and hdfs).
> Thread 1:
> {code}
> "pool-22-thread-2@4748" prio=5 tid=0x4d nid=NA waiting for monitor entry
> java.lang.Thread.State: BLOCKED
>waiting for pool-22-thread-1@4747 to release lock on <0x1423> (a
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader)
> at
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:91)
> at
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.loadClass(DelegatingClassLoader.java:367)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> at java.lang.Class.forName0(Class.java:-1)
> at java.lang.Class.forName(Class.java:348)
> at
> org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:715)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.enrich(ConnectorConfig.java:295)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:200)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:194)
> at
> org.apache.kafka.connect.runtime.Worker.startConnector(Worker.java:233)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.startConnector(DistributedHerder.java:916)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.access$1300(DistributedHerder.java:111)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:932)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:928)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:748)
> {code}
> Thread 2:
> {code}
> "pool-22-thread-1@4747" prio=5 tid=0x4c nid=NA waiting for monitor entry
> java.lang.Thread.State: BLOCKED
>blocks pool-22-thread-2@4748
>waiting for pool-22-thread-2@4748 to release lock on <0x1421> (a
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:406)
> at
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.loadClass(DelegatingClassLoader.java:358)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:411)
> - locked <0x1424> (a java.lang.Object)
> at
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:104)
> - locked <0x1423> (a
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> at
> io.debezium.transforms.ByLogicalTableRouter.(ByLogicalTableRouter.java:57)
> at java.lang.Class.forName0(Class.java:-1)
> at java.lang.Class.forName(Class.java:348)
> at
> org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:715)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.enrich(ConnectorConfig.java:295)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:200)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:194)
> at
> org.apache.kafka.connect.runtime.Worker.startConnector(Worker.java:233)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.startConnector(DistributedHerder.java:916)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.access$1300(DistributedHerder.java:111)
> at
>
[jira] [Commented] (KAFKA-7216) Exception while running kafka-acls.sh from 1.0 env on target Kafka env with 1.1.1
[
https://issues.apache.org/jira/browse/KAFKA-7216?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16624155#comment-16624155
]
ASF GitHub Bot commented on KAFKA-7216:
---
junrao closed pull request #5673: KAFKA-7216: Ignore unknown ResourceTypes
while loading acl cache
URL: https://github.com/apache/kafka/pull/5673
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala
b/core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala
index 7ec572caa66..892377cfe92 100644
--- a/core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala
+++ b/core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala
@@ -22,7 +22,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock
import com.typesafe.scalalogging.Logger
import kafka.api.KAFKA_2_0_IV1
import kafka.network.RequestChannel.Session
-import kafka.security.auth.SimpleAclAuthorizer.{VersionedAcls, NoAcls}
+import kafka.security.auth.SimpleAclAuthorizer.{NoAcls, VersionedAcls}
import kafka.server.KafkaConfig
import kafka.utils.CoreUtils.{inReadLock, inWriteLock}
import kafka.utils._
@@ -33,7 +33,7 @@ import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.apache.kafka.common.utils.{SecurityUtils, Time}
import scala.collection.JavaConverters._
-import scala.util.Random
+import scala.util.{Failure, Random, Success, Try}
object SimpleAclAuthorizer {
//optional override zookeeper cluster configuration where acls will be
stored, if not specified acls will be stored in
@@ -267,12 +267,17 @@ class SimpleAclAuthorizer extends Authorizer with Logging
{
ZkAclStore.stores.foreach(store => {
val resourceTypes = zkClient.getResourceTypes(store.patternType)
for (rType <- resourceTypes) {
- val resourceType = ResourceType.fromString(rType)
- val resourceNames = zkClient.getResourceNames(store.patternType,
resourceType)
- for (resourceName <- resourceNames) {
-val resource = new Resource(resourceType, resourceName,
store.patternType)
-val versionedAcls = getAclsFromZk(resource)
-updateCache(resource, versionedAcls)
+ val resourceType = Try(ResourceType.fromString(rType))
+ resourceType match {
+case Success(resourceTypeObj) => {
+ val resourceNames = zkClient.getResourceNames(store.patternType,
resourceTypeObj)
+ for (resourceName <- resourceNames) {
+val resource = new Resource(resourceTypeObj, resourceName,
store.patternType)
+val versionedAcls = getAclsFromZk(resource)
+updateCache(resource, versionedAcls)
+ }
+}
+case Failure(f) => warn(s"Ignoring unknown ResourceType: $rType")
}
}
})
This is an automated message from the Apache Git Service.
To respond to the message, please log on 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
> Exception while running kafka-acls.sh from 1.0 env on target Kafka env with
> 1.1.1
> -
>
> Key: KAFKA-7216
> URL: https://issues.apache.org/jira/browse/KAFKA-7216
> Project: Kafka
> Issue Type: Bug
> Components: admin
>Affects Versions: 1.1.0, 1.1.1, 2.0.0
>Reporter: Satish Duggana
>Assignee: Manikumar
>Priority: Major
>
> When `kafka-acls.sh` with SimpleAclAuthorizer on target Kafka cluster with
> 1.1.1 version, it throws the below error.
> {code:java}
> kafka.common.KafkaException: DelegationToken not a valid resourceType name.
> The valid names are Topic,Group,Cluster,TransactionalId
> at
> kafka.security.auth.ResourceType$$anonfun$fromString$1.apply(ResourceType.scala:56)
> at
> kafka.security.auth.ResourceType$$anonfun$fromString$1.apply(ResourceType.scala:56)
> at scala.Option.getOrElse(Option.scala:121)
> at kafka.security.auth.ResourceType$.fromString(ResourceType.scala:56)
> at
> kafka.security.auth.SimpleAclAuthorizer$$anonfun$loadCache$1$$anonfun$apply$mcV$sp$1.apply(SimpleAclAuthorizer.scala:233)
> at
> kafka.security.auth.SimpleAclAuthorizer$$anonfun$loadCache$1$$anonfun$apply$mcV$sp$1.apply(SimpleAclAuthorizer.scala:232)
> at scala.collection.Iterator$class.foreach(Iterator.scala:891)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
> at
[jira] [Commented] (KAFKA-7421) Deadlock in Kafka Connect
[
https://issues.apache.org/jira/browse/KAFKA-7421?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16624153#comment-16624153
]
Maciej Bryński commented on KAFKA-7421:
---
PS. I never have such a problem with older version of Kafka (confluent platform
4.0.0)
> Deadlock in Kafka Connect
> -
>
> Key: KAFKA-7421
> URL: https://issues.apache.org/jira/browse/KAFKA-7421
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
>Affects Versions: 2.0.0
>Reporter: Maciej Bryński
>Assignee: Konstantine Karantasis
>Priority: Critical
>
> I'm getting this deadlock on half of Kafka Connect runs when having two
> different types connectors (in this configuration it's debezium and hdfs).
> Thread 1:
> {code}
> "pool-22-thread-2@4748" prio=5 tid=0x4d nid=NA waiting for monitor entry
> java.lang.Thread.State: BLOCKED
>waiting for pool-22-thread-1@4747 to release lock on <0x1423> (a
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader)
> at
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:91)
> at
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.loadClass(DelegatingClassLoader.java:367)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> at java.lang.Class.forName0(Class.java:-1)
> at java.lang.Class.forName(Class.java:348)
> at
> org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:715)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.enrich(ConnectorConfig.java:295)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:200)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:194)
> at
> org.apache.kafka.connect.runtime.Worker.startConnector(Worker.java:233)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.startConnector(DistributedHerder.java:916)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.access$1300(DistributedHerder.java:111)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:932)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:928)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:748)
> {code}
> Thread 2:
> {code}
> "pool-22-thread-1@4747" prio=5 tid=0x4c nid=NA waiting for monitor entry
> java.lang.Thread.State: BLOCKED
>blocks pool-22-thread-2@4748
>waiting for pool-22-thread-2@4748 to release lock on <0x1421> (a
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:406)
> at
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.loadClass(DelegatingClassLoader.java:358)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:411)
> - locked <0x1424> (a java.lang.Object)
> at
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:104)
> - locked <0x1423> (a
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> at
> io.debezium.transforms.ByLogicalTableRouter.(ByLogicalTableRouter.java:57)
> at java.lang.Class.forName0(Class.java:-1)
> at java.lang.Class.forName(Class.java:348)
> at
> org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:715)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.enrich(ConnectorConfig.java:295)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:200)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:194)
> at
> org.apache.kafka.connect.runtime.Worker.startConnector(Worker.java:233)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.startConnector(DistributedHerder.java:916)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.access$1300(DistributedHerder.java:111)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:932)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:928)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
>
[jira] [Comment Edited] (KAFKA-7421) Deadlock in Kafka Connect
[
https://issues.apache.org/jira/browse/KAFKA-7421?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16624137#comment-16624137
]
Maciej Bryński edited comment on KAFKA-7421 at 9/21/18 8:35 PM:
[~kkonstantine] Nope. I was able to reproduce it when running connect process
with: -XX:+AlwaysLockClassLoader
was (Author: maver1ck):
[~kkonstantine] Nope. I was able to reproduce it wit running connect process
with: -XX:+AlwaysLockClassLoader
> Deadlock in Kafka Connect
> -
>
> Key: KAFKA-7421
> URL: https://issues.apache.org/jira/browse/KAFKA-7421
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
>Affects Versions: 2.0.0
>Reporter: Maciej Bryński
>Assignee: Konstantine Karantasis
>Priority: Critical
>
> I'm getting this deadlock on half of Kafka Connect runs when having two
> different types connectors (in this configuration it's debezium and hdfs).
> Thread 1:
> {code}
> "pool-22-thread-2@4748" prio=5 tid=0x4d nid=NA waiting for monitor entry
> java.lang.Thread.State: BLOCKED
>waiting for pool-22-thread-1@4747 to release lock on <0x1423> (a
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader)
> at
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:91)
> at
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.loadClass(DelegatingClassLoader.java:367)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> at java.lang.Class.forName0(Class.java:-1)
> at java.lang.Class.forName(Class.java:348)
> at
> org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:715)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.enrich(ConnectorConfig.java:295)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:200)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:194)
> at
> org.apache.kafka.connect.runtime.Worker.startConnector(Worker.java:233)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.startConnector(DistributedHerder.java:916)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.access$1300(DistributedHerder.java:111)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:932)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:928)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:748)
> {code}
> Thread 2:
> {code}
> "pool-22-thread-1@4747" prio=5 tid=0x4c nid=NA waiting for monitor entry
> java.lang.Thread.State: BLOCKED
>blocks pool-22-thread-2@4748
>waiting for pool-22-thread-2@4748 to release lock on <0x1421> (a
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:406)
> at
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.loadClass(DelegatingClassLoader.java:358)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:411)
> - locked <0x1424> (a java.lang.Object)
> at
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:104)
> - locked <0x1423> (a
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> at
> io.debezium.transforms.ByLogicalTableRouter.(ByLogicalTableRouter.java:57)
> at java.lang.Class.forName0(Class.java:-1)
> at java.lang.Class.forName(Class.java:348)
> at
> org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:715)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.enrich(ConnectorConfig.java:295)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:200)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:194)
> at
> org.apache.kafka.connect.runtime.Worker.startConnector(Worker.java:233)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.startConnector(DistributedHerder.java:916)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.access$1300(DistributedHerder.java:111)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:932)
> at
>
[jira] [Commented] (KAFKA-7421) Deadlock in Kafka Connect
[
https://issues.apache.org/jira/browse/KAFKA-7421?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16624137#comment-16624137
]
Maciej Bryński commented on KAFKA-7421:
---
[~kkonstantine] Nope. I was able to reproduce it wit running connect process
with: -XX:+AlwaysLockClassLoader
> Deadlock in Kafka Connect
> -
>
> Key: KAFKA-7421
> URL: https://issues.apache.org/jira/browse/KAFKA-7421
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
>Affects Versions: 2.0.0
>Reporter: Maciej Bryński
>Assignee: Konstantine Karantasis
>Priority: Critical
>
> I'm getting this deadlock on half of Kafka Connect runs when having two
> different types connectors (in this configuration it's debezium and hdfs).
> Thread 1:
> {code}
> "pool-22-thread-2@4748" prio=5 tid=0x4d nid=NA waiting for monitor entry
> java.lang.Thread.State: BLOCKED
>waiting for pool-22-thread-1@4747 to release lock on <0x1423> (a
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader)
> at
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:91)
> at
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.loadClass(DelegatingClassLoader.java:367)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> at java.lang.Class.forName0(Class.java:-1)
> at java.lang.Class.forName(Class.java:348)
> at
> org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:715)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.enrich(ConnectorConfig.java:295)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:200)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:194)
> at
> org.apache.kafka.connect.runtime.Worker.startConnector(Worker.java:233)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.startConnector(DistributedHerder.java:916)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.access$1300(DistributedHerder.java:111)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:932)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:928)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:748)
> {code}
> Thread 2:
> {code}
> "pool-22-thread-1@4747" prio=5 tid=0x4c nid=NA waiting for monitor entry
> java.lang.Thread.State: BLOCKED
>blocks pool-22-thread-2@4748
>waiting for pool-22-thread-2@4748 to release lock on <0x1421> (a
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:406)
> at
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.loadClass(DelegatingClassLoader.java:358)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:411)
> - locked <0x1424> (a java.lang.Object)
> at
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:104)
> - locked <0x1423> (a
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> at
> io.debezium.transforms.ByLogicalTableRouter.(ByLogicalTableRouter.java:57)
> at java.lang.Class.forName0(Class.java:-1)
> at java.lang.Class.forName(Class.java:348)
> at
> org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:715)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.enrich(ConnectorConfig.java:295)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:200)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:194)
> at
> org.apache.kafka.connect.runtime.Worker.startConnector(Worker.java:233)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.startConnector(DistributedHerder.java:916)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.access$1300(DistributedHerder.java:111)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:932)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:928)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
>
[jira] [Resolved] (KAFKA-7400) Compacted topic segments that precede the log start offset are not cleaned up
[ https://issues.apache.org/jira/browse/KAFKA-7400?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jun Rao resolved KAFKA-7400. Resolution: Fixed Fix Version/s: 2.1.0 Merged the PR to trunk. > Compacted topic segments that precede the log start offset are not cleaned up > - > > Key: KAFKA-7400 > URL: https://issues.apache.org/jira/browse/KAFKA-7400 > Project: Kafka > Issue Type: Bug > Components: log >Affects Versions: 1.1.0, 1.1.1, 2.0.0 >Reporter: Bob Barrett >Assignee: Bob Barrett >Priority: Minor > Fix For: 2.1.0 > > > LogManager.cleanupLogs currently checks if a topic is compacted, and skips > any deletion if it is. This means that if the log start offset increases, log > segments that precede the start offset will never be deleted. The log cleanup > logic should be improved to delete these segments even for compacted topics. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-7400) Compacted topic segments that precede the log start offset are not cleaned up
[
https://issues.apache.org/jira/browse/KAFKA-7400?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16624132#comment-16624132
]
ASF GitHub Bot commented on KAFKA-7400:
---
junrao closed pull request #5646: KAFKA-7400: Compacted topic segments that
precede the log start offse…
URL: https://github.com/apache/kafka/pull/5646
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/core/src/main/scala/kafka/log/Log.scala
b/core/src/main/scala/kafka/log/Log.scala
index afe151d69b6..c9b877bdca9 100644
--- a/core/src/main/scala/kafka/log/Log.scala
+++ b/core/src/main/scala/kafka/log/Log.scala
@@ -1353,12 +1353,17 @@ class Log(@volatile var dir: File,
}
/**
- * Delete any log segments that have either expired due to time based
retention
- * or because the log size is > retentionSize
+ * If topic deletion is enabled, delete any log segments that have either
expired due to time based retention
+ * or because the log size is > retentionSize.
+ *
+ * Whether or not deletion is enabled, delete any log segments that are
before the log start offset
*/
def deleteOldSegments(): Int = {
-if (!config.delete) return 0
-deleteRetentionMsBreachedSegments() +
deleteRetentionSizeBreachedSegments() + deleteLogStartOffsetBreachedSegments()
+if (config.delete) {
+ deleteRetentionMsBreachedSegments() +
deleteRetentionSizeBreachedSegments() + deleteLogStartOffsetBreachedSegments()
+} else {
+ deleteLogStartOffsetBreachedSegments()
+}
}
private def deleteRetentionMsBreachedSegments(): Int = {
diff --git a/core/src/main/scala/kafka/log/LogCleanerManager.scala
b/core/src/main/scala/kafka/log/LogCleanerManager.scala
index 83d902f952a..680fa94e33e 100755
--- a/core/src/main/scala/kafka/log/LogCleanerManager.scala
+++ b/core/src/main/scala/kafka/log/LogCleanerManager.scala
@@ -171,12 +171,13 @@ private[log] class LogCleanerManager(val logDirs:
Seq[File],
}
/**
-* Find any logs that have compact and delete enabled
+* Find any logs that have compaction enabled. Include logs without delete
enabled, as they may have segments
+* that precede the start offset.
*/
def deletableLogs(): Iterable[(TopicPartition, Log)] = {
inLock(lock) {
val toClean = logs.filter { case (topicPartition, log) =>
-!inProgress.contains(topicPartition) && isCompactAndDelete(log)
+!inProgress.contains(topicPartition) && log.config.compact
}
toClean.foreach { case (tp, _) => inProgress.put(tp,
LogCleaningInProgress) }
toClean
diff --git a/core/src/test/scala/unit/kafka/log/LogCleanerManagerTest.scala
b/core/src/test/scala/unit/kafka/log/LogCleanerManagerTest.scala
index 8cb2f9ec874..3653e282383 100644
--- a/core/src/test/scala/unit/kafka/log/LogCleanerManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogCleanerManagerTest.scala
@@ -77,17 +77,17 @@ class LogCleanerManagerTest extends JUnitSuite with Logging
{
}
/**
-* When looking for logs with segments ready to be deleted we shouldn't
consider
-* logs with cleanup.policy=compact as they shouldn't have segments
truncated.
+* When looking for logs with segments ready to be deleted we should
consider
+* logs with cleanup.policy=compact because they may have segments from
before the log start offset
*/
@Test
- def testLogsWithSegmentsToDeleteShouldNotConsiderCleanupPolicyCompactLogs():
Unit = {
+ def testLogsWithSegmentsToDeleteShouldConsiderCleanupPolicyCompactLogs():
Unit = {
val records = TestUtils.singletonRecords("test".getBytes,
key="test".getBytes)
val log: Log = createLog(records.sizeInBytes * 5, LogConfig.Compact)
val cleanerManager: LogCleanerManager = createCleanerManager(log)
val readyToDelete = cleanerManager.deletableLogs().size
-assertEquals("should have 1 logs ready to be deleted", 0, readyToDelete)
+assertEquals("should have 1 logs ready to be deleted", 1, readyToDelete)
}
/**
diff --git a/core/src/test/scala/unit/kafka/log/LogManagerTest.scala
b/core/src/test/scala/unit/kafka/log/LogManagerTest.scala
index 38d6f71c7d0..ae8bc01fab2 100755
--- a/core/src/test/scala/unit/kafka/log/LogManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogManagerTest.scala
@@ -190,13 +190,26 @@ class LogManagerTest {
}
/**
-* Ensures that LogManager only runs on logs with cleanup.policy=delete
+* Ensures that LogManager doesn't run on logs with
cleanup.policy=compact,delete
* LogCleaner.CleanerThread handles all logs where compaction is enabled.
*/
@Test
def testDoesntCleanLogsWithCompactDeletePolicy() {
+testDoesntCleanLogs(LogConfig.Compact + "," +
[jira] [Commented] (KAFKA-7430) Improve Transformer interface JavaDoc
[ https://issues.apache.org/jira/browse/KAFKA-7430?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16624043#comment-16624043 ] ASF GitHub Bot commented on KAFKA-7430: --- efeller opened a new pull request #5675: KAFKA-7430: Improve Transformer interface JavaDoc URL: https://github.com/apache/kafka/pull/5675 This PR improves the JavaDoc of the transformer interface. See https://issues.apache.org/jira/browse/KAFKA-7430 for more details. This is an automated message from the Apache Git Service. To respond to the message, please log on 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 > Improve Transformer interface JavaDoc > - > > Key: KAFKA-7430 > URL: https://issues.apache.org/jira/browse/KAFKA-7430 > Project: Kafka > Issue Type: Improvement > Components: streams >Affects Versions: 0.10.2.2, 0.11.0.3, 1.1.1, 2.0.0 >Reporter: Eugen Feller >Assignee: Eugen Feller >Priority: Trivial > Labels: stream > > Currently Transformer JavaDoc mentions that it is possible to use both > ctx.forward() and returning a KeyValue(). It would be great if we could > mention that returning a KeyValue is merely a convenience thing. In other > words, everything can be achieved using ctx.forward(): > "return new KeyValue()" and "ctx.forward(); return null;" are equivalent. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Assigned] (KAFKA-7430) Improve Transformer interface JavaDoc
[ https://issues.apache.org/jira/browse/KAFKA-7430?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Matthias J. Sax reassigned KAFKA-7430: -- Assignee: Eugen Feller > Improve Transformer interface JavaDoc > - > > Key: KAFKA-7430 > URL: https://issues.apache.org/jira/browse/KAFKA-7430 > Project: Kafka > Issue Type: Improvement > Components: streams >Affects Versions: 0.10.2.2, 0.11.0.3, 1.1.1, 2.0.0 >Reporter: Eugen Feller >Assignee: Eugen Feller >Priority: Trivial > Labels: stream > > Currently Transformer JavaDoc mentions that it is possible to use both > ctx.forward() and returning a KeyValue(). It would be great if we could > mention that returning a KeyValue is merely a convenience thing. In other > words, everything can be achieved using ctx.forward(): > "return new KeyValue()" and "ctx.forward(); return null;" are equivalent. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (KAFKA-7430) Improve Transformer interface JavaDoc
Eugen Feller created KAFKA-7430: --- Summary: Improve Transformer interface JavaDoc Key: KAFKA-7430 URL: https://issues.apache.org/jira/browse/KAFKA-7430 Project: Kafka Issue Type: Improvement Components: streams Affects Versions: 2.0.0, 1.1.1, 0.11.0.3, 0.10.2.2 Reporter: Eugen Feller Currently Transformer JavaDoc mentions that it is possible to use both ctx.forward() and returning a KeyValue(). It would be great if we could mention that returning a KeyValue is merely a convenience thing. In other words, everything can be achieved using ctx.forward(): "return new KeyValue()" and "ctx.forward(); return null;" are equivalent. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Assigned] (KAFKA-7216) Exception while running kafka-acls.sh from 1.0 env on target Kafka env with 1.1.1
[
https://issues.apache.org/jira/browse/KAFKA-7216?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Manikumar reassigned KAFKA-7216:
Assignee: Manikumar
> Exception while running kafka-acls.sh from 1.0 env on target Kafka env with
> 1.1.1
> -
>
> Key: KAFKA-7216
> URL: https://issues.apache.org/jira/browse/KAFKA-7216
> Project: Kafka
> Issue Type: Bug
> Components: admin
>Affects Versions: 1.1.0, 1.1.1, 2.0.0
>Reporter: Satish Duggana
>Assignee: Manikumar
>Priority: Major
>
> When `kafka-acls.sh` with SimpleAclAuthorizer on target Kafka cluster with
> 1.1.1 version, it throws the below error.
> {code:java}
> kafka.common.KafkaException: DelegationToken not a valid resourceType name.
> The valid names are Topic,Group,Cluster,TransactionalId
> at
> kafka.security.auth.ResourceType$$anonfun$fromString$1.apply(ResourceType.scala:56)
> at
> kafka.security.auth.ResourceType$$anonfun$fromString$1.apply(ResourceType.scala:56)
> at scala.Option.getOrElse(Option.scala:121)
> at kafka.security.auth.ResourceType$.fromString(ResourceType.scala:56)
> at
> kafka.security.auth.SimpleAclAuthorizer$$anonfun$loadCache$1$$anonfun$apply$mcV$sp$1.apply(SimpleAclAuthorizer.scala:233)
> at
> kafka.security.auth.SimpleAclAuthorizer$$anonfun$loadCache$1$$anonfun$apply$mcV$sp$1.apply(SimpleAclAuthorizer.scala:232)
> at scala.collection.Iterator$class.foreach(Iterator.scala:891)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
> at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
> at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
> at
> kafka.security.auth.SimpleAclAuthorizer$$anonfun$loadCache$1.apply$mcV$sp(SimpleAclAuthorizer.scala:232)
> at
> kafka.security.auth.SimpleAclAuthorizer$$anonfun$loadCache$1.apply(SimpleAclAuthorizer.scala:230)
> at
> kafka.security.auth.SimpleAclAuthorizer$$anonfun$loadCache$1.apply(SimpleAclAuthorizer.scala:230)
> at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:216)
> at kafka.utils.CoreUtils$.inWriteLock(CoreUtils.scala:224)
> at
> kafka.security.auth.SimpleAclAuthorizer.loadCache(SimpleAclAuthorizer.scala:230)
> at
> kafka.security.auth.SimpleAclAuthorizer.configure(SimpleAclAuthorizer.scala:114)
> at kafka.admin.AclCommand$.withAuthorizer(AclCommand.scala:83)
> at kafka.admin.AclCommand$.addAcl(AclCommand.scala:93)
> at kafka.admin.AclCommand$.main(AclCommand.scala:53)
> at kafka.admin.AclCommand.main(AclCommand.scala)
> {code}
>
> This is because it tries to get all the resource types registered from ZK
> path and it throws error when `DelegationToken` resource is not defined in
> `ResourceType` of client's Kafka version(which is earlier than 1.1.x)
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-7216) Exception while running kafka-acls.sh from 1.0 env on target Kafka env with 1.1.1
[
https://issues.apache.org/jira/browse/KAFKA-7216?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16623868#comment-16623868
]
ASF GitHub Bot commented on KAFKA-7216:
---
omkreddy opened a new pull request #5673: KAFKA-7216: Ignore unknown
ResourceTypes while loading acl cache
URL: https://github.com/apache/kafka/pull/5673
### 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 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
> Exception while running kafka-acls.sh from 1.0 env on target Kafka env with
> 1.1.1
> -
>
> Key: KAFKA-7216
> URL: https://issues.apache.org/jira/browse/KAFKA-7216
> Project: Kafka
> Issue Type: Bug
> Components: admin
>Affects Versions: 1.1.0, 1.1.1, 2.0.0
>Reporter: Satish Duggana
>Priority: Major
>
> When `kafka-acls.sh` with SimpleAclAuthorizer on target Kafka cluster with
> 1.1.1 version, it throws the below error.
> {code:java}
> kafka.common.KafkaException: DelegationToken not a valid resourceType name.
> The valid names are Topic,Group,Cluster,TransactionalId
> at
> kafka.security.auth.ResourceType$$anonfun$fromString$1.apply(ResourceType.scala:56)
> at
> kafka.security.auth.ResourceType$$anonfun$fromString$1.apply(ResourceType.scala:56)
> at scala.Option.getOrElse(Option.scala:121)
> at kafka.security.auth.ResourceType$.fromString(ResourceType.scala:56)
> at
> kafka.security.auth.SimpleAclAuthorizer$$anonfun$loadCache$1$$anonfun$apply$mcV$sp$1.apply(SimpleAclAuthorizer.scala:233)
> at
> kafka.security.auth.SimpleAclAuthorizer$$anonfun$loadCache$1$$anonfun$apply$mcV$sp$1.apply(SimpleAclAuthorizer.scala:232)
> at scala.collection.Iterator$class.foreach(Iterator.scala:891)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
> at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
> at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
> at
> kafka.security.auth.SimpleAclAuthorizer$$anonfun$loadCache$1.apply$mcV$sp(SimpleAclAuthorizer.scala:232)
> at
> kafka.security.auth.SimpleAclAuthorizer$$anonfun$loadCache$1.apply(SimpleAclAuthorizer.scala:230)
> at
> kafka.security.auth.SimpleAclAuthorizer$$anonfun$loadCache$1.apply(SimpleAclAuthorizer.scala:230)
> at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:216)
> at kafka.utils.CoreUtils$.inWriteLock(CoreUtils.scala:224)
> at
> kafka.security.auth.SimpleAclAuthorizer.loadCache(SimpleAclAuthorizer.scala:230)
> at
> kafka.security.auth.SimpleAclAuthorizer.configure(SimpleAclAuthorizer.scala:114)
> at kafka.admin.AclCommand$.withAuthorizer(AclCommand.scala:83)
> at kafka.admin.AclCommand$.addAcl(AclCommand.scala:93)
> at kafka.admin.AclCommand$.main(AclCommand.scala:53)
> at kafka.admin.AclCommand.main(AclCommand.scala)
> {code}
>
> This is because it tries to get all the resource types registered from ZK
> path and it throws error when `DelegationToken` resource is not defined in
> `ResourceType` of client's Kafka version(which is earlier than 1.1.x)
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Resolved] (KAFKA-7425) Kafka Broker -Unable to start
[ https://issues.apache.org/jira/browse/KAFKA-7425?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Manikumar resolved KAFKA-7425. -- Resolution: Duplicate will handle this as part of KAFKA-7216. > Kafka Broker -Unable to start > -- > > Key: KAFKA-7425 > URL: https://issues.apache.org/jira/browse/KAFKA-7425 > Project: Kafka > Issue Type: Bug >Affects Versions: 1.0.0 >Reporter: Sathish Yanamala >Priority: Blocker > > Hello Team, > We are facing below Error, While starting Kafka Broker . > By disabling below property in server.properties , we can able to up Broker , > But we are missing ACL authentication with Kafka Brokers. > ## > server.properties (Config File) > authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer > > > Please help and suggest on below error . > Error Log : > 2018-09-20 13:50:54,121 INFO > kafka.coordinator.transaction.TransactionCoordinator: [TransactionCoordinator > id=4] Startup complete. > 2018-09-20 13:50:54,238 FATAL kafka.server.KafkaServer: [KafkaServer id=4] > Fatal error during KafkaServer startup. Prepare to shutdown > kafka.common.KafkaException: DelegationToken not a valid resourceType name. > The valid names are Topic,Group,Cluster,TransactionalId > at > kafka.security.auth.ResourceType$.$anonfun$fromString$2(ResourceType.scala:56) > at scala.Option.getOrElse(Option.scala:121) > at kafka.security.auth.ResourceType$.fromString(ResourceType.scala:56) > at > kafka.security.auth.SimpleAclAuthorizer.$anonfun$loadCache$2(SimpleAclAuthorizer.scala:233) > at > kafka.security.auth.SimpleAclAuthorizer.$anonfun$loadCache$2$adapted(SimpleAclAuthorizer.scala:232) > at scala.collection.Iterator.foreach(Iterator.scala:929) > at scala.collection.Iterator.foreach$(Iterator.scala:929) > at scala.collection.AbstractIterator.foreach(Iterator.scala:1417) > at scala.collection.IterableLike.foreach(IterableLike.scala:71) > at scala.collection.IterableLike.foreach$(IterableLike.scala:70) > at scala.collection.AbstractIterable.foreach(Iterable.scala:54) > at > kafka.security.auth.SimpleAclAuthorizer.$anonfun$loadCache$1(SimpleAclAuthorizer.scala:232) > at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:12) > at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:217) > at kafka.utils.CoreUtils$.inWriteLock(CoreUtils.scala:225) > at > kafka.security.auth.SimpleAclAuthorizer.loadCache(SimpleAclAuthorizer.scala:230) > at > kafka.security.auth.SimpleAclAuthorizer.configure(SimpleAclAuthorizer.scala:114) > at kafka.server.KafkaServer.$anonfun$startup$4(KafkaServer.scala:254) > at scala.Option.map(Option.scala:146) > at kafka.server.KafkaServer.startup(KafkaServer.scala:252) > at kafka.server.KafkaServerStartable.startup(KafkaServerStartable.scala:38) > at kafka.Kafka$.main(Kafka.scala:92) > at kafka.Kafka.main(Kafka.scala) > 2018-09-20 13:50:54,241 INFO kafka.server.KafkaServer: [KafkaServer id=4] > shutting down > 2018-09-20 13:50:54,242 INFO kafka.network.SocketServer: [SocketServer > brokerId=4] Shutting down > 2018-09-20 13:50:54,260 INFO kafka.network.SocketServer: [SocketServer > brokerId=4] Shutdown completed > 2018-09-20 13:50:54,267 INFO > kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper: > [ExpirationReaper-4-topic]: Shutting down > 2018-09-20 13:50:54,451 INFO > kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper: > [ExpirationReaper-4-topic]: Shutdown completed > 2018-09-20 13:50:54,451 INFO > kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper: > [ExpirationReaper-4-topic]: Stopped > 2018-09-20 13:50:54,453 INFO > kafka.coordinator.transaction.TransactionCoordinator: [TransactionCoordinator > id=4] Shutting down. > 2018-09-20 13:50:54,454 INFO kafka.coordinator.transaction.ProducerIdManager: > [ProducerId Manager 4]: Shutdown complete: last producerId assigned 1559000 > 2018-09-20 13:50:54,455 INFO > kafka.coordinator.transaction.TransactionStateManager: [Transaction State > Manager 4]: Shutdown complete > 2018-09-20 13:50:54,455 INFO > kafka.coordinator.transaction.TransactionMarkerChannelManager: [Transaction > Marker Channel Manager 4]: Shutting down > 2018-09-20 13:50:54,455 INFO > kafka.coordinator.transaction.TransactionMarkerChannelManager: [Transaction > Marker Channel Manager 4]: Stopped > 2018-09-20 13:50:54,455 INFO > kafka.coordinator.transaction.TransactionMarkerChannelManager: [Transaction > Marker Channel Manager 4]: Shutdown completed > 2018-09-20 13:50:54,456 INFO > kafka.coordinator.transaction.TransactionCoordinator: [TransactionCoordinator > id=4] Shutdown complete. > 2018-09-20 13:50:54,456 INFO kafka.coordinator.group.GroupCoordinator: > [GroupCoordinator 4]: Shutting down. > 2018-09-20 13:50:54,457 INFO >
[jira] [Commented] (KAFKA-3268) Refactor existing CLI scripts to use KafkaAdminClient
[ https://issues.apache.org/jira/browse/KAFKA-3268?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16623611#comment-16623611 ] Eno Thereska commented on KAFKA-3268: - Is there an update on these? If not, might be good to release so others in the community can participate. > Refactor existing CLI scripts to use KafkaAdminClient > - > > Key: KAFKA-3268 > URL: https://issues.apache.org/jira/browse/KAFKA-3268 > Project: Kafka > Issue Type: Sub-task >Reporter: Grant Henke >Assignee: Viktor Somogyi >Priority: Major > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-5561) Refactor TopicCommand to use the Admin client
[ https://issues.apache.org/jira/browse/KAFKA-5561?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16623460#comment-16623460 ] Viktor Somogyi commented on KAFKA-5561: --- [~ppatierno] are you currently working on this? I've seen the last modification was around last august and I'd like to pick this up if you don't mind. > Refactor TopicCommand to use the Admin client > - > > Key: KAFKA-5561 > URL: https://issues.apache.org/jira/browse/KAFKA-5561 > Project: Kafka > Issue Type: Improvement > Components: tools >Reporter: Paolo Patierno >Assignee: Paolo Patierno >Priority: Major > > Hi, > as suggested in the https://issues.apache.org/jira/browse/KAFKA-3331, it > could be great to have the TopicCommand using the new Admin client instead of > the way it works today. > As pushed by [~gwenshap] in the above JIRA, I'm going to work on it. > Thanks, > Paolo -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-1774) REPL and Shell Client for Admin Message RQ/RP
[ https://issues.apache.org/jira/browse/KAFKA-1774?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16623300#comment-16623300 ] Viktor Somogyi commented on KAFKA-1774: --- [~ijuma], [~junrao] do you think this is still required? I have an implementation which works for topics (add, list, describe, delete), configs (set/update, delete, describe), consumergroups (describe, list), log dirs (describe, list). Are you interested in a demo? > REPL and Shell Client for Admin Message RQ/RP > - > > Key: KAFKA-1774 > URL: https://issues.apache.org/jira/browse/KAFKA-1774 > Project: Kafka > Issue Type: Sub-task >Reporter: Joe Stein >Assignee: Viktor Somogyi >Priority: Major > > We should have a REPL we can work in and execute the commands with the > arguments. With this we can do: > ./kafka.sh --shell > kafka>attach cluster -b localhost:9092; > kafka>describe topic sampleTopicNameForExample; > the command line version can work like it does now so folks don't have to > re-write all of their tooling. > kafka.sh --topics --everything the same like kafka-topics.sh is > kafka.sh --reassign --everything the same like kafka-reassign-partitions.sh > is -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (KAFKA-7429) Enable dynamic key/truststore update with same filename/password
Rajini Sivaram created KAFKA-7429: - Summary: Enable dynamic key/truststore update with same filename/password Key: KAFKA-7429 URL: https://issues.apache.org/jira/browse/KAFKA-7429 Project: Kafka Issue Type: Improvement Components: security Reporter: Rajini Sivaram Assignee: Rajini Sivaram Fix For: 2.1.0 At the moment, SSL keystores and truststores on brokers can be dynamically updated using AdminClient by providing a new keystore or truststore. But we require either the filename or password to be modified to trigger the update. In some scenarios, we may want to perform the update using the same file (and password). So it will be good to provide a way to trigger reload of existing keystores and truststores using the same AdminClient update mechanism. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
