[jira] [Commented] (KAFKA-4682) Committed offsets should not be deleted if a consumer is still active (KIP-211)
[
https://issues.apache.org/jira/browse/KAFKA-4682?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16519864#comment-16519864
]
ASF GitHub Bot commented on KAFKA-4682:
---
hachikuji closed pull request #4896: KAFKA-4682: Revise expiration semantics of
consumer group offsets (KIP-211 - Part 1)
URL: https://github.com/apache/kafka/pull/4896
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/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
index 9c19af17037..b484e110aee 100644
---
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
+++
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
@@ -798,8 +798,7 @@ public void onComplete(Map offsets, Exception
OffsetCommitRequest.Builder builder = new
OffsetCommitRequest.Builder(this.groupId, offsetData).
setGenerationId(generation.generationId).
-setMemberId(generation.memberId).
-setRetentionTime(OffsetCommitRequest.DEFAULT_RETENTION_TIME);
+setMemberId(generation.memberId);
log.trace("Sending OffsetCommit request with {} to coordinator {}",
offsets, coordinator);
diff --git
a/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java
b/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java
index 570c4d5a66a..8a51e84e76a 100644
---
a/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java
+++
b/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java
@@ -111,9 +111,15 @@
*/
private static final Schema OFFSET_COMMIT_REQUEST_V4 =
OFFSET_COMMIT_REQUEST_V3;
+private static final Schema OFFSET_COMMIT_REQUEST_V5 = new Schema(
+GROUP_ID,
+GENERATION_ID,
+MEMBER_ID,
+new Field(TOPICS_KEY_NAME, new
ArrayOf(OFFSET_COMMIT_REQUEST_TOPIC_V2), "Topics to commit offsets."));
+
public static Schema[] schemaVersions() {
return new Schema[] {OFFSET_COMMIT_REQUEST_V0,
OFFSET_COMMIT_REQUEST_V1, OFFSET_COMMIT_REQUEST_V2,
-OFFSET_COMMIT_REQUEST_V3, OFFSET_COMMIT_REQUEST_V4};
+OFFSET_COMMIT_REQUEST_V3, OFFSET_COMMIT_REQUEST_V4,
OFFSET_COMMIT_REQUEST_V5};
}
// default values for the current version
@@ -166,7 +172,6 @@ public String toString() {
private final Map offsetData;
private String memberId = DEFAULT_MEMBER_ID;
private int generationId = DEFAULT_GENERATION_ID;
-private long retentionTime = DEFAULT_RETENTION_TIME;
public Builder(String groupId, Map
offsetData) {
super(ApiKeys.OFFSET_COMMIT);
@@ -184,11 +189,6 @@ public Builder setGenerationId(int generationId) {
return this;
}
-public Builder setRetentionTime(long retentionTime) {
-this.retentionTime = retentionTime;
-return this;
-}
-
@Override
public OffsetCommitRequest build(short version) {
switch (version) {
@@ -199,8 +199,8 @@ public OffsetCommitRequest build(short version) {
case 2:
case 3:
case 4:
-long retentionTime = version == 1 ? DEFAULT_RETENTION_TIME
: this.retentionTime;
-return new OffsetCommitRequest(groupId, generationId,
memberId, retentionTime, offsetData, version);
+case 5:
+return new OffsetCommitRequest(groupId, generationId,
memberId, DEFAULT_RETENTION_TIME, offsetData, version);
default:
throw new UnsupportedVersionException("Unsupported version
" + version);
}
@@ -213,7 +213,6 @@ public String toString() {
append(", groupId=").append(groupId).
append(", memberId=").append(memberId).
append(", generationId=").append(generationId).
-append(", retentionTime=").append(retentionTime).
append(", offsetData=").append(offsetData).
append(")");
return bld.toString();
@@ -316,6 +315,7 @@ public AbstractResponse getErrorResponse(int
throttleTimeMs, Throwable e) {
return new OffsetCommitResponse(responseData);
case 3:
case 4:
+case 5:
return new OffsetCommitResponse(throttleTimeMs, responseData);
default:
throw new IllegalArgumentExceptio
[jira] [Resolved] (KAFKA-7082) Concurrent createTopics calls may throw NodeExistsException
[
https://issues.apache.org/jira/browse/KAFKA-7082?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ismael Juma resolved KAFKA-7082.
Resolution: Fixed
Reviewer: Jun Rao
> Concurrent createTopics calls may throw NodeExistsException
> ---
>
> Key: KAFKA-7082
> URL: https://issues.apache.org/jira/browse/KAFKA-7082
> Project: Kafka
> Issue Type: Bug
>Affects Versions: 1.1.0
>Reporter: Ismael Juma
>Assignee: Ismael Juma
>Priority: Critical
> Labels: regression
> Fix For: 2.0.1, 1.1.2
>
>
> This exception is unexpected causing an `UnknownServerException` to be thrown
> back to the client. Example below:
> {code}
> org.apache.zookeeper.KeeperException$NodeExistsException: KeeperErrorCode =
> NodeExists for /config/topics/connect-configs
> at
> org.apache.zookeeper.KeeperException.create(KeeperException.java:119)
> at
> org.apache.zookeeper.KeeperException.create(KeeperException.java:51)
> at kafka.zookeeper.AsyncResponse.maybeThrow(ZooKeeperClient.scala:472)
> at kafka.zk.KafkaZkClient.createRecursive(KafkaZkClient.scala:1400)
> at kafka.zk.KafkaZkClient.create$1(KafkaZkClient.scala:262)
> at
> kafka.zk.KafkaZkClient.setOrCreateEntityConfigs(KafkaZkClient.scala:269)
> at
> kafka.zk.AdminZkClient.createOrUpdateTopicPartitionAssignmentPathInZK(AdminZkClient.scala:99)
> at kafka.server.AdminManager$$anonfun$2.apply(AdminManager.scala:126)
> at kafka.server.AdminManager$$anonfun$2.apply(AdminManager.scala:81)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
> 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
> scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-7082) Concurrent createTopics calls may throw NodeExistsException
[
https://issues.apache.org/jira/browse/KAFKA-7082?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16519850#comment-16519850
]
ASF GitHub Bot commented on KAFKA-7082:
---
ijuma closed pull request #5259: KAFKA-7082: Concurrent create topics may throw
NodeExistsException
URL: https://github.com/apache/kafka/pull/5259
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/zk/AdminZkClient.scala
b/core/src/main/scala/kafka/zk/AdminZkClient.scala
index 8a6b3ee212d..060c0b4d4ae 100644
--- a/core/src/main/scala/kafka/zk/AdminZkClient.scala
+++ b/core/src/main/scala/kafka/zk/AdminZkClient.scala
@@ -93,7 +93,6 @@ class AdminZkClient(zkClient: KafkaZkClient) extends Logging {
update: Boolean = false) {
validateCreateOrUpdateTopic(topic, partitionReplicaAssignment, config,
update)
-// Configs only matter if a topic is being created. Changing configs via
AlterTopic is not supported
if (!update) {
// write out the config if there is any, this isn't transactional with
the partition assignments
zkClient.setOrCreateEntityConfigs(ConfigType.Topic, topic, config)
diff --git a/core/src/main/scala/kafka/zk/KafkaZkClient.scala
b/core/src/main/scala/kafka/zk/KafkaZkClient.scala
index bb342945ea8..ec4932ab47b 100644
--- a/core/src/main/scala/kafka/zk/KafkaZkClient.scala
+++ b/core/src/main/scala/kafka/zk/KafkaZkClient.scala
@@ -246,6 +246,12 @@ class KafkaZkClient private (zooKeeperClient:
ZooKeeperClient, isSecure: Boolean
/**
* Sets or creates the entity znode path with the given configs depending
* on whether it already exists or not.
+ *
+ * If this is method is called concurrently, the last writer wins. In cases
where we update configs and then
+ * partition assignment (i.e. create topic), it's possible for one thread to
set this and the other to set the
+ * partition assignment. As such, the recommendation is to never call create
topic for the same topic with different
+ * configs/partition assignment concurrently.
+ *
* @param rootEntityType entity type
* @param sanitizedEntityName entity name
* @throws KeeperException if there is an error while setting or creating
the znode
@@ -257,16 +263,19 @@ class KafkaZkClient private (zooKeeperClient:
ZooKeeperClient, isSecure: Boolean
retryRequestUntilConnected(setDataRequest)
}
-def create(configData: Array[Byte]) = {
+def createOrSet(configData: Array[Byte]): Unit = {
val path = ConfigEntityZNode.path(rootEntityType, sanitizedEntityName)
- createRecursive(path, ConfigEntityZNode.encode(config))
+ try createRecursive(path, ConfigEntityZNode.encode(config))
+ catch {
+case _: NodeExistsException => set(configData).maybeThrow
+ }
}
val configData = ConfigEntityZNode.encode(config)
val setDataResponse = set(configData)
setDataResponse.resultCode match {
- case Code.NONODE => create(configData)
+ case Code.NONODE => createOrSet(configData)
case _ => setDataResponse.maybeThrow
}
}
diff --git a/core/src/test/scala/unit/kafka/zk/AdminZkClientTest.scala
b/core/src/test/scala/unit/kafka/zk/AdminZkClientTest.scala
index fe5fbff55d0..39745e5e608 100644
--- a/core/src/test/scala/unit/kafka/zk/AdminZkClientTest.scala
+++ b/core/src/test/scala/unit/kafka/zk/AdminZkClientTest.scala
@@ -28,8 +28,10 @@ import kafka.utils.TestUtils._
import kafka.utils.{Logging, TestUtils}
import kafka.zk.{AdminZkClient, KafkaZkClient, ZooKeeperTestHarness}
import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.config.TopicConfig
import org.apache.kafka.common.errors.{InvalidReplicaAssignmentException,
InvalidTopicException, TopicExistsException}
import org.apache.kafka.common.metrics.Quota
+import org.apache.kafka.test.{TestUtils => JTestUtils}
import org.easymock.EasyMock
import org.junit.Assert._
import org.junit.{After, Test}
@@ -132,7 +134,7 @@ class AdminZkClientTest extends ZooKeeperTestHarness with
Logging with RackAware
}
@Test
- def testConcurrentTopicCreation() {
+ def testMockedConcurrentTopicCreation() {
val topic = "test.topic"
// simulate the ZK interactions that can happen when a topic is
concurrently created by multiple processes
@@ -147,6 +149,28 @@ class AdminZkClientTest extends ZooKeeperTestHarness with
Logging with RackAware
}
}
+ @Test
+ def testConcurrentTopicCreation() {
+val topic = "test-concurrent-topic-creation"
+TestUtils.createBrokersInZk(zkClient, List(0, 1, 2, 3, 4))
+val props = new Properties
+props.setProperty(TopicConfi
[jira] [Commented] (KAFKA-7089) Extend `kafka-consumer-groups.sh` to show "beginning offsets"
[ https://issues.apache.org/jira/browse/KAFKA-7089?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16519849#comment-16519849 ] Matthias J. Sax commented on KAFKA-7089: AFAIK, tools are considered public API, so I guess yes. The reason is, that some people build pipelines that parse the output of the command line tools (think, grep, sed, etc) and we need to make sure we don't break anything. \cc [~hachikuji] WDYT? > Extend `kafka-consumer-groups.sh` to show "beginning offsets" > - > > Key: KAFKA-7089 > URL: https://issues.apache.org/jira/browse/KAFKA-7089 > Project: Kafka > Issue Type: Improvement > Components: tools >Reporter: Matthias J. Sax >Assignee: Vahid Hashemian >Priority: Minor > > Currently, `kafka-consumer-groups.sh` only shows "current offset", "end > offset" and "lag". It would be helpful to extend the tool to also show > "beginning/earliest offset". -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-7089) Extend `kafka-consumer-groups.sh` to show "beginning offsets"
[ https://issues.apache.org/jira/browse/KAFKA-7089?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16519837#comment-16519837 ] Vahid Hashemian commented on KAFKA-7089: [~mjsax], do you think a KIP is required for this? > Extend `kafka-consumer-groups.sh` to show "beginning offsets" > - > > Key: KAFKA-7089 > URL: https://issues.apache.org/jira/browse/KAFKA-7089 > Project: Kafka > Issue Type: Improvement > Components: tools >Reporter: Matthias J. Sax >Assignee: Vahid Hashemian >Priority: Minor > > Currently, `kafka-consumer-groups.sh` only shows "current offset", "end > offset" and "lag". It would be helpful to extend the tool to also show > "beginning/earliest offset". -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-5886) Introduce delivery.timeout.ms producer config (KIP-91)
[ https://issues.apache.org/jira/browse/KAFKA-5886?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16519833#comment-16519833 ] ASF GitHub Bot commented on KAFKA-5886: --- yuyang08 opened a new pull request #5270: KAFKA-5886: Introduce delivery.timeout.ms producer config (KIP-91) URL: https://github.com/apache/kafka/pull/5270 This change is based on @sutambe 's change https://github.com/apache/kafka/pull/3849 earlier. primary changes in this pr: 1. In RecordAccumulator.java, use `inFlightBatches` to track the in-flight batches, instead of using `soonToExpireInFlightsBatches` to only track the soon-to-expire batches. With this change, in RecordAccumulator.expiredBatches, we check both `inFlightBatches` and `batches` to find the expired batches. 2. Fixed the test failures in SenderTest.java and RecordAccumulatorTest.java. 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 > Introduce delivery.timeout.ms producer config (KIP-91) > -- > > Key: KAFKA-5886 > URL: https://issues.apache.org/jira/browse/KAFKA-5886 > Project: Kafka > Issue Type: Improvement > Components: producer >Reporter: Sumant Tambe >Assignee: Sumant Tambe >Priority: Major > > We propose adding a new timeout delivery.timeout.ms. The window of > enforcement includes batching in the accumulator, retries, and the inflight > segments of the batch. With this config, the user has a guaranteed upper > bound on when a record will either get sent, fail or expire from the point > when send returns. In other words we no longer overload request.timeout.ms to > act as a weak proxy for accumulator timeout and instead introduce an explicit > timeout that users can rely on without exposing any internals of the producer > such as the accumulator. > See > [KIP-91|https://cwiki.apache.org/confluence/display/KAFKA/KIP-91+Provide+Intuitive+User+Timeouts+in+The+Producer] > for more details. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-7041) Using RocksDB bulk loading for StandbyTasks
[ https://issues.apache.org/jira/browse/KAFKA-7041?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16519827#comment-16519827 ] Matthias J. Sax commented on KAFKA-7041: Feel free to dig into this. On a high level, if should be sufficient to open RocksDB with the correct settings for StandbyTask. Atm, when a store is created, we always open it with regular setting, and if we detect that restore is required, we close, change configs, reopen, restore, close, change configs back, reopen, start processing. I think, for StandbyTask we can actually be smarter and always open them via bulk-load configs directly. I don't think there is much more we need to consider. Thus, we need to set RocksDBStore into a different "state" (maybe some kind of flag) for StandbyTasks. Hope this helps to get you started :) > Using RocksDB bulk loading for StandbyTasks > --- > > Key: KAFKA-7041 > URL: https://issues.apache.org/jira/browse/KAFKA-7041 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: Matthias J. Sax >Priority: Major > > In KAFKA-5363 we introduced RocksDB bulk loading to speed up store recovery. > We could do the same optimization for StandbyTasks to make them more > efficient and to reduce the likelihood that StandbyTasks lag behind. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-7041) Using RocksDB bulk loading for StandbyTasks
[ https://issues.apache.org/jira/browse/KAFKA-7041?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16519812#comment-16519812 ] Nikki Thean commented on KAFKA-7041: [~mjsax] Thank you for taking the time to explain! I think I understand now; I had assumed that [this|https://github.com/apache/kafka/blob/d3e264e773c4652f34b40c7c3b494c0f7fbabffc/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java#L284] meant bulk loading, but now that I have looked at the code for WriteBatch in RocksDB I see that is not actually what 'batch' meant in this case and that it's necessary to toggle the DB configs for actual bulk loading. I'm still interested in this ticket if you think of any neat way to deal with the standby task issue (I'd need to read the code again and learn a bit more about what RocksDB actually does during bulk loading before I could come close to a suggestion myself). Let me know about the windowed stores too. > Using RocksDB bulk loading for StandbyTasks > --- > > Key: KAFKA-7041 > URL: https://issues.apache.org/jira/browse/KAFKA-7041 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: Matthias J. Sax >Priority: Major > > In KAFKA-5363 we introduced RocksDB bulk loading to speed up store recovery. > We could do the same optimization for StandbyTasks to make them more > efficient and to reduce the likelihood that StandbyTasks lag behind. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Assigned] (KAFKA-6556) allow to attach callbacks in kafka streams, to be triggered when a window is expired
[ https://issues.apache.org/jira/browse/KAFKA-6556?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] John Roesler reassigned KAFKA-6556: --- Assignee: John Roesler > allow to attach callbacks in kafka streams, to be triggered when a window is > expired > - > > Key: KAFKA-6556 > URL: https://issues.apache.org/jira/browse/KAFKA-6556 > Project: Kafka > Issue Type: New Feature > Components: streams >Reporter: igor mazor >Assignee: John Roesler >Priority: Major > > Allowing to attach callbacks in kafka streams, to be triggered when a window > is expired, > would help in use cases when the final state of the window is required. > It would be also useful if together with that functionally the user would be > able to control whether the callback would be triggered in addition to > emitting the normal change log down the stream, or only triggering the > callback when the window is expired. (for example in cases when only the > final window state is required, and any updates to the window state during > the window time interval are not important) > An example for use case could be left join with proper sql semantics: > A left join B currently would emit (a1,null) and (a1,b1) if b1 arrives > within the defined join time window. > what I would like is to have ONLY ONE result: > (a1,null) if no b1 arrived during the defined join time window, OR ONLY > (a1,b1) if b1 did arrived in the specified join time window. > One possible solution could be to use the current kafka streams left join > with downstream processor which would put the results in a windowed Ktable. > The window size would be same as for the left join operation, however, only > the final state of that window would be emitted down the stream once the time > window is expired. > So if the left join produces (a1, null) and after X minutes no (a1, b1) was > produced, eventually only (a1, null) would be emitted, on the other hand, if > the left join produces (a1, null) and after X-t minutes (a1, b1) was produced > by the left join operation => only (a1, b1) would be emitted eventually down > the stream after X minutes. > > Another use case is when the window state is written to another kafka topic > which is then used to persist the window states into a db, However, many > times only the final window state > is required, and functionality to get only the last window state would help > in reducing load from the db, since only the final window state would be > persisted to the db, instead of multiple db writes for each window state > update. > > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-7012) Performance issue upgrading to kafka 1.0.1 or 1.1
[ https://issues.apache.org/jira/browse/KAFKA-7012?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16519694#comment-16519694 ] Corey Kasten commented on KAFKA-7012: - On behalf of [~rajadayalanvdms] and myself, thank you all! > Performance issue upgrading to kafka 1.0.1 or 1.1 > - > > Key: KAFKA-7012 > URL: https://issues.apache.org/jira/browse/KAFKA-7012 > Project: Kafka > Issue Type: Bug >Affects Versions: 1.1.0, 1.0.1 >Reporter: rajadayalan perumalsamy >Assignee: Rajini Sivaram >Priority: Critical > Labels: regression > Fix For: 2.0.0, 1.0.2, 1.1.1 > > Attachments: Commit-47ee8e954-0607-bufferkeys-nopoll-profile.png, > Commit-47ee8e954-0607-memory.png, Commit-47ee8e954-0607-profile.png, > Commit-47ee8e954-profile.png, Commit-47ee8e954-profile2.png, > Commit-f15cdbc91b-profile.png, Commit-f15cdbc91b-profile2.png > > > We are trying to upgrade kafka cluster from Kafka 0.11.0.1 to Kafka 1.0.1. > After upgrading 1 node on the cluster, we notice that network threads use > most of the cpu. It is a 3 node cluster with 15k messages/sec on each node. > With Kafka 0.11.0.1 typical usage of the servers is around 50 to 60% > vcpu(using less than 1 vcpu). After upgrade we are noticing that cpu usage is > high depending on the number of network threads used. If networks threads is > set to 8, then the cpu usage is around 850%(9 vcpus) and if it is set to 4 > then the cpu usage is around 450%(5 vcpus). Using the same kafka > server.properties for both. > Did further analysis with git bisect, couple of build and deploys, traced the > issue to commit 47ee8e954df62b9a79099e944ec4be29afe046f6. CPU usage is fine > for commit f15cdbc91b240e656d9a2aeb6877e94624b21f8d. But with commit > 47ee8e954df62b9a79099e944ec4be29afe046f6 cpu usage has increased. Have > attached screenshots of profiling done with both the commits. Screenshot > Commit-f15cdbc91b-profile shows less cpu usage by network threads and > Screenshots Commit-47ee8e954-profile and Commit-47ee8e954-profile2 show > higher cpu usage(almost entire cpu usage) by network threads. Also noticed > that kafka.network.Processor.poll() method is invoked 10 times more with > commit 47ee8e954df62b9a79099e944ec4be29afe046f6. > We need the issue to be resolved to upgrade the cluster. Please let me know > if you need any additional information. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-7012) Performance issue upgrading to kafka 1.0.1 or 1.1
[ https://issues.apache.org/jira/browse/KAFKA-7012?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16519666#comment-16519666 ] Ismael Juma commented on KAFKA-7012: Dong has published 1.1.1 RC1 due to another issue which means that this will be in 1.1.1. > Performance issue upgrading to kafka 1.0.1 or 1.1 > - > > Key: KAFKA-7012 > URL: https://issues.apache.org/jira/browse/KAFKA-7012 > Project: Kafka > Issue Type: Bug >Affects Versions: 1.1.0, 1.0.1 >Reporter: rajadayalan perumalsamy >Assignee: Rajini Sivaram >Priority: Critical > Labels: regression > Fix For: 2.0.0, 1.0.2, 1.1.1 > > Attachments: Commit-47ee8e954-0607-bufferkeys-nopoll-profile.png, > Commit-47ee8e954-0607-memory.png, Commit-47ee8e954-0607-profile.png, > Commit-47ee8e954-profile.png, Commit-47ee8e954-profile2.png, > Commit-f15cdbc91b-profile.png, Commit-f15cdbc91b-profile2.png > > > We are trying to upgrade kafka cluster from Kafka 0.11.0.1 to Kafka 1.0.1. > After upgrading 1 node on the cluster, we notice that network threads use > most of the cpu. It is a 3 node cluster with 15k messages/sec on each node. > With Kafka 0.11.0.1 typical usage of the servers is around 50 to 60% > vcpu(using less than 1 vcpu). After upgrade we are noticing that cpu usage is > high depending on the number of network threads used. If networks threads is > set to 8, then the cpu usage is around 850%(9 vcpus) and if it is set to 4 > then the cpu usage is around 450%(5 vcpus). Using the same kafka > server.properties for both. > Did further analysis with git bisect, couple of build and deploys, traced the > issue to commit 47ee8e954df62b9a79099e944ec4be29afe046f6. CPU usage is fine > for commit f15cdbc91b240e656d9a2aeb6877e94624b21f8d. But with commit > 47ee8e954df62b9a79099e944ec4be29afe046f6 cpu usage has increased. Have > attached screenshots of profiling done with both the commits. Screenshot > Commit-f15cdbc91b-profile shows less cpu usage by network threads and > Screenshots Commit-47ee8e954-profile and Commit-47ee8e954-profile2 show > higher cpu usage(almost entire cpu usage) by network threads. Also noticed > that kafka.network.Processor.poll() method is invoked 10 times more with > commit 47ee8e954df62b9a79099e944ec4be29afe046f6. > We need the issue to be resolved to upgrade the cluster. Please let me know > if you need any additional information. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-7012) Performance issue upgrading to kafka 1.0.1 or 1.1
[ https://issues.apache.org/jira/browse/KAFKA-7012?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ismael Juma updated KAFKA-7012: --- Fix Version/s: (was: 1.1.2) 1.1.1 > Performance issue upgrading to kafka 1.0.1 or 1.1 > - > > Key: KAFKA-7012 > URL: https://issues.apache.org/jira/browse/KAFKA-7012 > Project: Kafka > Issue Type: Bug >Affects Versions: 1.1.0, 1.0.1 >Reporter: rajadayalan perumalsamy >Assignee: Rajini Sivaram >Priority: Critical > Labels: regression > Fix For: 2.0.0, 1.0.2, 1.1.1 > > Attachments: Commit-47ee8e954-0607-bufferkeys-nopoll-profile.png, > Commit-47ee8e954-0607-memory.png, Commit-47ee8e954-0607-profile.png, > Commit-47ee8e954-profile.png, Commit-47ee8e954-profile2.png, > Commit-f15cdbc91b-profile.png, Commit-f15cdbc91b-profile2.png > > > We are trying to upgrade kafka cluster from Kafka 0.11.0.1 to Kafka 1.0.1. > After upgrading 1 node on the cluster, we notice that network threads use > most of the cpu. It is a 3 node cluster with 15k messages/sec on each node. > With Kafka 0.11.0.1 typical usage of the servers is around 50 to 60% > vcpu(using less than 1 vcpu). After upgrade we are noticing that cpu usage is > high depending on the number of network threads used. If networks threads is > set to 8, then the cpu usage is around 850%(9 vcpus) and if it is set to 4 > then the cpu usage is around 450%(5 vcpus). Using the same kafka > server.properties for both. > Did further analysis with git bisect, couple of build and deploys, traced the > issue to commit 47ee8e954df62b9a79099e944ec4be29afe046f6. CPU usage is fine > for commit f15cdbc91b240e656d9a2aeb6877e94624b21f8d. But with commit > 47ee8e954df62b9a79099e944ec4be29afe046f6 cpu usage has increased. Have > attached screenshots of profiling done with both the commits. Screenshot > Commit-f15cdbc91b-profile shows less cpu usage by network threads and > Screenshots Commit-47ee8e954-profile and Commit-47ee8e954-profile2 show > higher cpu usage(almost entire cpu usage) by network threads. Also noticed > that kafka.network.Processor.poll() method is invoked 10 times more with > commit 47ee8e954df62b9a79099e944ec4be29afe046f6. > We need the issue to be resolved to upgrade the cluster. Please let me know > if you need any additional information. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Assigned] (KAFKA-7089) Extend `kafka-consumer-groups.sh` to show "beginning offsets"
[ https://issues.apache.org/jira/browse/KAFKA-7089?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Vahid Hashemian reassigned KAFKA-7089: -- Assignee: Vahid Hashemian > Extend `kafka-consumer-groups.sh` to show "beginning offsets" > - > > Key: KAFKA-7089 > URL: https://issues.apache.org/jira/browse/KAFKA-7089 > Project: Kafka > Issue Type: Improvement > Components: tools >Reporter: Matthias J. Sax >Assignee: Vahid Hashemian >Priority: Minor > > Currently, `kafka-consumer-groups.sh` only shows "current offset", "end > offset" and "lag". It would be helpful to extend the tool to also show > "beginning/earliest offset". -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (KAFKA-7089) Extend `kafka-consumer-groups.sh` to show "beginning offsets"
Matthias J. Sax created KAFKA-7089: -- Summary: Extend `kafka-consumer-groups.sh` to show "beginning offsets" Key: KAFKA-7089 URL: https://issues.apache.org/jira/browse/KAFKA-7089 Project: Kafka Issue Type: Improvement Components: tools Reporter: Matthias J. Sax Currently, `kafka-consumer-groups.sh` only shows "current offset", "end offset" and "lag". It would be helpful to extend the tool to also show "beginning/earliest offset". -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Assigned] (KAFKA-7075) Allow Topology#addGlobalStore to add a window store
[
https://issues.apache.org/jira/browse/KAFKA-7075?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Guozhang Wang reassigned KAFKA-7075:
Assignee: Nishanth Pradeep
> Allow Topology#addGlobalStore to add a window store
> ---
>
> Key: KAFKA-7075
> URL: https://issues.apache.org/jira/browse/KAFKA-7075
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Reporter: Guozhang Wang
>Assignee: Nishanth Pradeep
>Priority: Major
> Labels: newbie
>
> Today although {{Topology#addGlobalStore}} can take any {{StateStore}} types,
> the internal implementation {{InternalTopologyBuilder#addGlobalStore}} only
> accepts {{StoreBuilder}}. It means if users pass in a windowed
> store builder in {{Topology#addGlobalStore}} it will cause a runtime
> ClassCastException.
> We should fix this issue by relaxing the {{InternalTopologyBuilder}}.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-7075) Allow Topology#addGlobalStore to add a window store
[
https://issues.apache.org/jira/browse/KAFKA-7075?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16519601#comment-16519601
]
Guozhang Wang commented on KAFKA-7075:
--
Thanks [~nprad]. I've added you as the assignee.
> Allow Topology#addGlobalStore to add a window store
> ---
>
> Key: KAFKA-7075
> URL: https://issues.apache.org/jira/browse/KAFKA-7075
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Reporter: Guozhang Wang
>Priority: Major
> Labels: newbie
>
> Today although {{Topology#addGlobalStore}} can take any {{StateStore}} types,
> the internal implementation {{InternalTopologyBuilder#addGlobalStore}} only
> accepts {{StoreBuilder}}. It means if users pass in a windowed
> store builder in {{Topology#addGlobalStore}} it will cause a runtime
> ClassCastException.
> We should fix this issue by relaxing the {{InternalTopologyBuilder}}.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-7088) Kafka streams thread waits infinitely on transaction init
[
https://issues.apache.org/jira/browse/KAFKA-7088?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16519600#comment-16519600
]
Guozhang Wang commented on KAFKA-7088:
--
[~lgluchowski] The blocking code inside producer is:
{code}
if (initTransactionsResult.await(maxBlockTimeMs,
TimeUnit.MILLISECONDS)) {
initTransactionsResult = null;
}
{code}
The {{maxBlockTimeMs}} is defined as {{ProducerConfig.MAX_BLOCK_MS_CONFIG}}, by
default it is 60 seconds. Did you override this config? Also note that under
normal network condition and broker health, producer should wait much less than
60 seconds to get the result from broker. Did you observe any broker
unavailability or network issues during then?
> Kafka streams thread waits infinitely on transaction init
> -
>
> Key: KAFKA-7088
> URL: https://issues.apache.org/jira/browse/KAFKA-7088
> Project: Kafka
> Issue Type: Bug
> Components: clients
>Affects Versions: 1.0.1
> Environment: Linux 4.14.33-51.37.amzn1.x86_64 #1 SMP Thu May 3
> 20:07:43 UTC 2018
> kafka-streams (client) 1.0.1
> kafka broker 1.1.0
> Java version:
> OpenJDK Runtime Environment (build 1.8.0_171-b10)
> OpenJDK 64-Bit Server VM (build 25.171-b10, mixed mode)
> kakfa config overrides:
> num.stream.threads: 6
> session.timeout.ms: 1
> request.timeout.ms: 11000
> fetch.max.wait.ms: 500
> max.poll.records: 1000
> topic has 24 partitions
>Reporter: Lukasz Gluchowski
>Priority: Major
> Labels: eos
>
> A kafka stream application thread stops processing without any feedback. The
> topic has 24 partitions and I noticed that processing stopped only for some
> partitions. I will describe what happened to partition:10. The application is
> still running (now for about 8 hours) and that thread is hanging there and no
> rebalancing that took place.
> There is no error (we have a custom `Thread.UncaughtExceptionHandler` which
> was not called). I noticed that after couple of minutes stream stopped
> processing (at offset 32606948 where log-end-offset is 33472402).
> Broker itself is not reporting any active consumer in that consumer group and
> the only info I was able to gather was from thread dump:
> {code:java}
> "mp_ads_publisher_pro_madstorage-web-corotos-prod-9db804ae-2a7a-431f-be09-392ab38cd8a2-StreamThread-33"
> #113 prio=5 os_prio=0 tid=0x7fe07c6349b0 nid=0xf7a waiting on condition
> [0x7fe0215d4000]
> java.lang.Thread.State: WAITING (parking)
> at sun.misc.Unsafe.park(Native Method)
> - parking to wait for <0xfec6a2f8> (a
> java.util.concurrent.CountDownLatch$Sync)
> at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
> at
> java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836)
> at
> java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997)
> at
> java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304)
> at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231)
> at
> org.apache.kafka.clients.producer.internals.TransactionalRequestResult.await(TransactionalRequestResult.java:50)
> at
> org.apache.kafka.clients.producer.KafkaProducer.initTransactions(KafkaProducer.java:554)
> at
> org.apache.kafka.streams.processor.internals.StreamTask.(StreamTask.java:151)
> at
> org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:404)
> at
> org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:365)
> at
> org.apache.kafka.streams.processor.internals.StreamThread$AbstractTaskCreator.createTasks(StreamThread.java:350)
> at
> org.apache.kafka.streams.processor.internals.TaskManager.addStreamTasks(TaskManager.java:137)
> at
> org.apache.kafka.streams.processor.internals.TaskManager.createTasks(TaskManager.java:88)
> at
> org.apache.kafka.streams.processor.internals.StreamThread$RebalanceListener.onPartitionsAssigned(StreamThread.java:259)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:264)
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:367)
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:316)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:295)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1146)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:)
> at
> org.apache.k
[jira] [Commented] (KAFKA-7079) ValueTransformer#transform does not pass the key
[ https://issues.apache.org/jira/browse/KAFKA-7079?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16519589#comment-16519589 ] Hashan Gayasri Udugahapattuwa commented on KAFKA-7079: -- [~mjsax] will do that. Thanks again for the quick responses . > ValueTransformer#transform does not pass the key > > > Key: KAFKA-7079 > URL: https://issues.apache.org/jira/browse/KAFKA-7079 > Project: Kafka > Issue Type: Bug > Components: streams >Affects Versions: 1.1.0 > Environment: Fedora 27 >Reporter: Hashan Gayasri Udugahapattuwa >Priority: Major > > ValueTransformers' transform method doesn't pass the key to user-code. > Reporting this as a bug since it currently requires workarounds. > > Context: > I'm currently in the process of converting two stateful "*aggregate*" DSL > operations to the Processor API since the state of those operations are > relatively large and takes 99% + of CPU time (when profiled) for serializing > and deserializing them via Kryo. > Since DSL aggregations use state stores of [Bytes, Array[Byte]]] even when > using the in-memory state store, it seems like the only way to reduce the > serialization/deserialization overhead is to convert heavy aggregates to > *transform*s. > In my case, *ValueTransformer* seems to be the option. However, since > ValueTransformers' _transform_ method only exposes the _value_, I'd either > have to pre-process and add the key to the value or use *Transformer* instead > (which is not my intent). > > As internal _*InternalValueTransformerWithKey*_ already has the readOnlyKey, > it seems like a good idea to pass the key to the transform method as well, > esp since in a stateful transformation, generally the state store has to be > queried by the key. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-7079) ValueTransformer#transform does not pass the key
[ https://issues.apache.org/jira/browse/KAFKA-7079?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16519575#comment-16519575 ] Matthias J. Sax commented on KAFKA-7079: [~hashan] We can leave this ticket open and update it to track the missing docs. Can you update the Jira title and description accordingly? Thanks. > ValueTransformer#transform does not pass the key > > > Key: KAFKA-7079 > URL: https://issues.apache.org/jira/browse/KAFKA-7079 > Project: Kafka > Issue Type: Bug > Components: streams >Affects Versions: 1.1.0 > Environment: Fedora 27 >Reporter: Hashan Gayasri Udugahapattuwa >Priority: Major > > ValueTransformers' transform method doesn't pass the key to user-code. > Reporting this as a bug since it currently requires workarounds. > > Context: > I'm currently in the process of converting two stateful "*aggregate*" DSL > operations to the Processor API since the state of those operations are > relatively large and takes 99% + of CPU time (when profiled) for serializing > and deserializing them via Kryo. > Since DSL aggregations use state stores of [Bytes, Array[Byte]]] even when > using the in-memory state store, it seems like the only way to reduce the > serialization/deserialization overhead is to convert heavy aggregates to > *transform*s. > In my case, *ValueTransformer* seems to be the option. However, since > ValueTransformers' _transform_ method only exposes the _value_, I'd either > have to pre-process and add the key to the value or use *Transformer* instead > (which is not my intent). > > As internal _*InternalValueTransformerWithKey*_ already has the readOnlyKey, > it seems like a good idea to pass the key to the transform method as well, > esp since in a stateful transformation, generally the state store has to be > queried by the key. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-7088) Kafka streams thread waits infinitely on transaction init
[
https://issues.apache.org/jira/browse/KAFKA-7088?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Matthias J. Sax updated KAFKA-7088:
---
Labels: eos (was: )
> Kafka streams thread waits infinitely on transaction init
> -
>
> Key: KAFKA-7088
> URL: https://issues.apache.org/jira/browse/KAFKA-7088
> Project: Kafka
> Issue Type: Bug
> Components: clients
>Affects Versions: 1.0.1
> Environment: Linux 4.14.33-51.37.amzn1.x86_64 #1 SMP Thu May 3
> 20:07:43 UTC 2018
> kafka-streams (client) 1.0.1
> kafka broker 1.1.0
> Java version:
> OpenJDK Runtime Environment (build 1.8.0_171-b10)
> OpenJDK 64-Bit Server VM (build 25.171-b10, mixed mode)
> kakfa config overrides:
> num.stream.threads: 6
> session.timeout.ms: 1
> request.timeout.ms: 11000
> fetch.max.wait.ms: 500
> max.poll.records: 1000
> topic has 24 partitions
>Reporter: Lukasz Gluchowski
>Priority: Major
> Labels: eos
>
> A kafka stream application thread stops processing without any feedback. The
> topic has 24 partitions and I noticed that processing stopped only for some
> partitions. I will describe what happened to partition:10. The application is
> still running (now for about 8 hours) and that thread is hanging there and no
> rebalancing that took place.
> There is no error (we have a custom `Thread.UncaughtExceptionHandler` which
> was not called). I noticed that after couple of minutes stream stopped
> processing (at offset 32606948 where log-end-offset is 33472402).
> Broker itself is not reporting any active consumer in that consumer group and
> the only info I was able to gather was from thread dump:
> {code:java}
> "mp_ads_publisher_pro_madstorage-web-corotos-prod-9db804ae-2a7a-431f-be09-392ab38cd8a2-StreamThread-33"
> #113 prio=5 os_prio=0 tid=0x7fe07c6349b0 nid=0xf7a waiting on condition
> [0x7fe0215d4000]
> java.lang.Thread.State: WAITING (parking)
> at sun.misc.Unsafe.park(Native Method)
> - parking to wait for <0xfec6a2f8> (a
> java.util.concurrent.CountDownLatch$Sync)
> at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
> at
> java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836)
> at
> java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997)
> at
> java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304)
> at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231)
> at
> org.apache.kafka.clients.producer.internals.TransactionalRequestResult.await(TransactionalRequestResult.java:50)
> at
> org.apache.kafka.clients.producer.KafkaProducer.initTransactions(KafkaProducer.java:554)
> at
> org.apache.kafka.streams.processor.internals.StreamTask.(StreamTask.java:151)
> at
> org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:404)
> at
> org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:365)
> at
> org.apache.kafka.streams.processor.internals.StreamThread$AbstractTaskCreator.createTasks(StreamThread.java:350)
> at
> org.apache.kafka.streams.processor.internals.TaskManager.addStreamTasks(TaskManager.java:137)
> at
> org.apache.kafka.streams.processor.internals.TaskManager.createTasks(TaskManager.java:88)
> at
> org.apache.kafka.streams.processor.internals.StreamThread$RebalanceListener.onPartitionsAssigned(StreamThread.java:259)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:264)
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:367)
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:316)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:295)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1146)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:851)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:808)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:774)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:744){code}
>
> I tried restarting application once but the situation repeated. Thread read
> some data, committed offset and stopped processing, leaving that thread in
> wait state.
> FYI: we have EOS enabled
--
This message was sen
[jira] [Commented] (KAFKA-7088) Kafka streams thread waits infinitely on transaction init
[
https://issues.apache.org/jira/browse/KAFKA-7088?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16519572#comment-16519572
]
Matthias J. Sax commented on KAFKA-7088:
According to the stack trace it hangs in the producer. Thus, it seems to be
client issues, not a Kafka Streams issues. \cc [~hachikuji]
> Kafka streams thread waits infinitely on transaction init
> -
>
> Key: KAFKA-7088
> URL: https://issues.apache.org/jira/browse/KAFKA-7088
> Project: Kafka
> Issue Type: Bug
> Components: clients
>Affects Versions: 1.0.1
> Environment: Linux 4.14.33-51.37.amzn1.x86_64 #1 SMP Thu May 3
> 20:07:43 UTC 2018
> kafka-streams (client) 1.0.1
> kafka broker 1.1.0
> Java version:
> OpenJDK Runtime Environment (build 1.8.0_171-b10)
> OpenJDK 64-Bit Server VM (build 25.171-b10, mixed mode)
> kakfa config overrides:
> num.stream.threads: 6
> session.timeout.ms: 1
> request.timeout.ms: 11000
> fetch.max.wait.ms: 500
> max.poll.records: 1000
> topic has 24 partitions
>Reporter: Lukasz Gluchowski
>Priority: Major
>
> A kafka stream application thread stops processing without any feedback. The
> topic has 24 partitions and I noticed that processing stopped only for some
> partitions. I will describe what happened to partition:10. The application is
> still running (now for about 8 hours) and that thread is hanging there and no
> rebalancing that took place.
> There is no error (we have a custom `Thread.UncaughtExceptionHandler` which
> was not called). I noticed that after couple of minutes stream stopped
> processing (at offset 32606948 where log-end-offset is 33472402).
> Broker itself is not reporting any active consumer in that consumer group and
> the only info I was able to gather was from thread dump:
> {code:java}
> "mp_ads_publisher_pro_madstorage-web-corotos-prod-9db804ae-2a7a-431f-be09-392ab38cd8a2-StreamThread-33"
> #113 prio=5 os_prio=0 tid=0x7fe07c6349b0 nid=0xf7a waiting on condition
> [0x7fe0215d4000]
> java.lang.Thread.State: WAITING (parking)
> at sun.misc.Unsafe.park(Native Method)
> - parking to wait for <0xfec6a2f8> (a
> java.util.concurrent.CountDownLatch$Sync)
> at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
> at
> java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836)
> at
> java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997)
> at
> java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304)
> at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231)
> at
> org.apache.kafka.clients.producer.internals.TransactionalRequestResult.await(TransactionalRequestResult.java:50)
> at
> org.apache.kafka.clients.producer.KafkaProducer.initTransactions(KafkaProducer.java:554)
> at
> org.apache.kafka.streams.processor.internals.StreamTask.(StreamTask.java:151)
> at
> org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:404)
> at
> org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:365)
> at
> org.apache.kafka.streams.processor.internals.StreamThread$AbstractTaskCreator.createTasks(StreamThread.java:350)
> at
> org.apache.kafka.streams.processor.internals.TaskManager.addStreamTasks(TaskManager.java:137)
> at
> org.apache.kafka.streams.processor.internals.TaskManager.createTasks(TaskManager.java:88)
> at
> org.apache.kafka.streams.processor.internals.StreamThread$RebalanceListener.onPartitionsAssigned(StreamThread.java:259)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:264)
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:367)
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:316)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:295)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1146)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:851)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:808)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:774)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:744){code}
>
> I tried restarting application once but the situation repeated. Thread read
>
[jira] [Updated] (KAFKA-7088) Kafka streams thread waits infinitely on transaction init
[
https://issues.apache.org/jira/browse/KAFKA-7088?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Matthias J. Sax updated KAFKA-7088:
---
Component/s: (was: streams)
clients
> Kafka streams thread waits infinitely on transaction init
> -
>
> Key: KAFKA-7088
> URL: https://issues.apache.org/jira/browse/KAFKA-7088
> Project: Kafka
> Issue Type: Bug
> Components: clients
>Affects Versions: 1.0.1
> Environment: Linux 4.14.33-51.37.amzn1.x86_64 #1 SMP Thu May 3
> 20:07:43 UTC 2018
> kafka-streams (client) 1.0.1
> kafka broker 1.1.0
> Java version:
> OpenJDK Runtime Environment (build 1.8.0_171-b10)
> OpenJDK 64-Bit Server VM (build 25.171-b10, mixed mode)
> kakfa config overrides:
> num.stream.threads: 6
> session.timeout.ms: 1
> request.timeout.ms: 11000
> fetch.max.wait.ms: 500
> max.poll.records: 1000
> topic has 24 partitions
>Reporter: Lukasz Gluchowski
>Priority: Major
>
> A kafka stream application thread stops processing without any feedback. The
> topic has 24 partitions and I noticed that processing stopped only for some
> partitions. I will describe what happened to partition:10. The application is
> still running (now for about 8 hours) and that thread is hanging there and no
> rebalancing that took place.
> There is no error (we have a custom `Thread.UncaughtExceptionHandler` which
> was not called). I noticed that after couple of minutes stream stopped
> processing (at offset 32606948 where log-end-offset is 33472402).
> Broker itself is not reporting any active consumer in that consumer group and
> the only info I was able to gather was from thread dump:
> {code:java}
> "mp_ads_publisher_pro_madstorage-web-corotos-prod-9db804ae-2a7a-431f-be09-392ab38cd8a2-StreamThread-33"
> #113 prio=5 os_prio=0 tid=0x7fe07c6349b0 nid=0xf7a waiting on condition
> [0x7fe0215d4000]
> java.lang.Thread.State: WAITING (parking)
> at sun.misc.Unsafe.park(Native Method)
> - parking to wait for <0xfec6a2f8> (a
> java.util.concurrent.CountDownLatch$Sync)
> at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
> at
> java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836)
> at
> java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997)
> at
> java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304)
> at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231)
> at
> org.apache.kafka.clients.producer.internals.TransactionalRequestResult.await(TransactionalRequestResult.java:50)
> at
> org.apache.kafka.clients.producer.KafkaProducer.initTransactions(KafkaProducer.java:554)
> at
> org.apache.kafka.streams.processor.internals.StreamTask.(StreamTask.java:151)
> at
> org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:404)
> at
> org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:365)
> at
> org.apache.kafka.streams.processor.internals.StreamThread$AbstractTaskCreator.createTasks(StreamThread.java:350)
> at
> org.apache.kafka.streams.processor.internals.TaskManager.addStreamTasks(TaskManager.java:137)
> at
> org.apache.kafka.streams.processor.internals.TaskManager.createTasks(TaskManager.java:88)
> at
> org.apache.kafka.streams.processor.internals.StreamThread$RebalanceListener.onPartitionsAssigned(StreamThread.java:259)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:264)
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:367)
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:316)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:295)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1146)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:851)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:808)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:774)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:744){code}
>
> I tried restarting application once but the situation repeated. Thread read
> some data, committed offset and stopped processing, leaving that thread in
> wait state.
> FYI: we have EOS enabled
--
This messag
[jira] [Commented] (KAFKA-6977) Kafka Streams - java.lang.IllegalStateException: Unexpected error code 2 while fetching data
[
https://issues.apache.org/jira/browse/KAFKA-6977?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16519546#comment-16519546
]
Eugen Feller commented on KAFKA-6977:
-
[~guozhang] Cool. Makes sense. In my case, I wanted to avoid changing consumer
group as it would mean data loss. In that case having something like
[https://kafka.apache.org/10/documentation/streams/developer-guide/config-streams#default-deserialization-exception-handler]
for consumer would be great.
> Kafka Streams - java.lang.IllegalStateException: Unexpected error code 2
> while fetching data
> -
>
> Key: KAFKA-6977
> URL: https://issues.apache.org/jira/browse/KAFKA-6977
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 0.11.0.1
>Reporter: Eugen Feller
>Priority: Blocker
> Labels: streams
>
> We are running Kafka Streams 0.11.0.1 with Kafka Broker 0.10.2.1 and
> constantly run into the following exception:
> {code:java}
> [visitstats-mongowriter-838cf286-be42-4a49-b3ab-3a291617233e-StreamThread-1]
> INFO org.apache.kafka.streams.processor.internals.StreamThread -
> stream-thread
> [visitstats-mongowriter-838cf286-be42-4a49-b3ab-3a291617233e-StreamThread-1]
> partition assignment took 40 ms.
> current active tasks: [0_0, 0_1, 0_3, 0_4, 0_5, 0_10, 0_12, 0_13, 0_14, 0_15,
> 0_17, 0_20, 0_21, 0_23, 0_25, 0_28, 0_2, 0_6, 0_7, 0_8, 0_9, 0_11, 0_16,
> 0_18, 0_19, 0_22, 0_24, 0_26, 0_27, 0_29, 0_30, 0_31]
> current standby tasks: []
> previous active tasks: [0_0, 0_1, 0_3, 0_4, 0_5, 0_10, 0_12, 0_13, 0_14,
> 0_15, 0_17, 0_20, 0_21, 0_23, 0_25, 0_28]
> [visitstats-mongowriter-838cf286-be42-4a49-b3ab-3a291617233e-StreamThread-1]
> INFO org.apache.kafka.streams.processor.internals.StreamThread -
> stream-thread
> [visitstats-mongowriter-838cf286-be42-4a49-b3ab-3a291617233e-StreamThread-1]
> State transition from PARTITIONS_ASSIGNED to RUNNING.
> [visitstats-mongowriter-838cf286-be42-4a49-b3ab-3a291617233e-StreamThread-1]
> INFO org.apache.kafka.streams.KafkaStreams - stream-client
> [visitstats-mongowriter-838cf286-be42-4a49-b3ab-3a291617233e] State
> transition from REBALANCING to RUNNING.
> [visitstats-mongowriter-838cf286-be42-4a49-b3ab-3a291617233e-StreamThread-1]
> ERROR org.apache.kafka.streams.processor.internals.StreamThread -
> stream-thread
> [visitstats-mongowriter-838cf286-be42-4a49-b3ab-3a291617233e-StreamThread-1]
> Encountered the following error during processing:
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:480)
> at
> org.apache.kafka.clients.consumer.internals.Fetcher.parseCompletedFetch(Fetcher.java:886)
> at
> org.apache.kafka.clients.consumer.internals.Fetcher.fetchedRecords(Fetcher.java:525)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1086)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1043)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:536)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:490)
> java.lang.IllegalStateException: Unexpected error code 2 while fetching data
> at
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:457)
> [visitstats-mongowriter-3727734b-3c2f-4775-87d2-838d72ca3bf4-StreamThread-1]
> INFO org.apache.kafka.streams.processor.internals.StreamThread -
> stream-thread
> [visitstats-mongowriter-3727734b-3c2f-4775-87d2-838d72ca3bf4-StreamThread-1]
> Shutting down
> [visitstats-mongowriter-3727734b-3c2f-4775-87d2-838d72ca3bf4-StreamThread-1]
> INFO org.apache.kafka.streams.processor.internals.StreamThread -
> stream-thread
> [visitstats-mongowriter-3727734b-3c2f-4775-87d2-838d72ca3bf4-StreamThread-1]
> State transition from RUNNING to PENDING_SHUTDOWN.
> [visitstats-mongowriter-3727734b-3c2f-4775-87d2-838d72ca3bf4-StreamThread-1]
> INFO org.apache.kafka.clients.producer.KafkaProducer - Closing the Kafka
> producer with timeoutMillis = 9223372036854775807 ms.
> [visitstats-mongowriter-3727734b-3c2f-4775-87d2-838d72ca3bf4-StreamThread-1]
> INFO org.apache.kafka.streams.processor.internals.StreamThread -
> stream-thread
> [visitstats-mongowriter-3727734b-3c2f-4775-87d2-838d72ca3bf4-StreamThread-1]
> Stream thread shutdown complete
> [visitstats-mongowriter-3727734b-3c2f-4775-87d2-838d72ca3bf4-StreamThread-1]
> INFO org.apache.kafka.streams.processor.internals.StreamThread -
> stream-thread
> [visitstats-mongowriter-3727734b-3c2f-4775-87d2-838d72ca3bf4-StreamThread-1]
> State transition from PENDING_SHUTDOWN to DEAD.
> [visitstats-mongowriter-3727734b-3c2f-4775-87d2-838d72ca3bf4-StreamThread-1]
> INFO org.apache.kafka.stre
[jira] [Commented] (KAFKA-7088) Kafka streams thread waits infinitely on transaction init
[
https://issues.apache.org/jira/browse/KAFKA-7088?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16519542#comment-16519542
]
Ted Yu commented on KAFKA-7088:
---
Is it possible to reproduce this behavior using 1.1.1 RC ?
There have been some changes in related code since 1.0.1
Thanks
> Kafka streams thread waits infinitely on transaction init
> -
>
> Key: KAFKA-7088
> URL: https://issues.apache.org/jira/browse/KAFKA-7088
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 1.0.1
> Environment: Linux 4.14.33-51.37.amzn1.x86_64 #1 SMP Thu May 3
> 20:07:43 UTC 2018
> kafka-streams (client) 1.0.1
> kafka broker 1.1.0
> Java version:
> OpenJDK Runtime Environment (build 1.8.0_171-b10)
> OpenJDK 64-Bit Server VM (build 25.171-b10, mixed mode)
> kakfa config overrides:
> num.stream.threads: 6
> session.timeout.ms: 1
> request.timeout.ms: 11000
> fetch.max.wait.ms: 500
> max.poll.records: 1000
> topic has 24 partitions
>Reporter: Lukasz Gluchowski
>Priority: Major
>
> A kafka stream application thread stops processing without any feedback. The
> topic has 24 partitions and I noticed that processing stopped only for some
> partitions. I will describe what happened to partition:10. The application is
> still running (now for about 8 hours) and that thread is hanging there and no
> rebalancing that took place.
> There is no error (we have a custom `Thread.UncaughtExceptionHandler` which
> was not called). I noticed that after couple of minutes stream stopped
> processing (at offset 32606948 where log-end-offset is 33472402).
> Broker itself is not reporting any active consumer in that consumer group and
> the only info I was able to gather was from thread dump:
> {code:java}
> "mp_ads_publisher_pro_madstorage-web-corotos-prod-9db804ae-2a7a-431f-be09-392ab38cd8a2-StreamThread-33"
> #113 prio=5 os_prio=0 tid=0x7fe07c6349b0 nid=0xf7a waiting on condition
> [0x7fe0215d4000]
> java.lang.Thread.State: WAITING (parking)
> at sun.misc.Unsafe.park(Native Method)
> - parking to wait for <0xfec6a2f8> (a
> java.util.concurrent.CountDownLatch$Sync)
> at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
> at
> java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836)
> at
> java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997)
> at
> java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304)
> at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231)
> at
> org.apache.kafka.clients.producer.internals.TransactionalRequestResult.await(TransactionalRequestResult.java:50)
> at
> org.apache.kafka.clients.producer.KafkaProducer.initTransactions(KafkaProducer.java:554)
> at
> org.apache.kafka.streams.processor.internals.StreamTask.(StreamTask.java:151)
> at
> org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:404)
> at
> org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:365)
> at
> org.apache.kafka.streams.processor.internals.StreamThread$AbstractTaskCreator.createTasks(StreamThread.java:350)
> at
> org.apache.kafka.streams.processor.internals.TaskManager.addStreamTasks(TaskManager.java:137)
> at
> org.apache.kafka.streams.processor.internals.TaskManager.createTasks(TaskManager.java:88)
> at
> org.apache.kafka.streams.processor.internals.StreamThread$RebalanceListener.onPartitionsAssigned(StreamThread.java:259)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:264)
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:367)
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:316)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:295)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1146)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:851)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:808)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:774)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:744){code}
>
> I tried restarting application once but the situation repeated. Thread read
> some data, committed offset and
[jira] [Updated] (KAFKA-7088) Kafka streams thread waits infinitely on transaction init
[
https://issues.apache.org/jira/browse/KAFKA-7088?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Lukasz Gluchowski updated KAFKA-7088:
-
Description:
A kafka stream application thread stops processing without any feedback. The
topic has 24 partitions and I noticed that processing stopped only for some
partitions. I will describe what happened to partition:10. The application is
still running (now for about 8 hours) and that thread is hanging there and no
rebalancing that took place.
There is no error (we have a custom `Thread.UncaughtExceptionHandler` which was
not called). I noticed that after couple of minutes stream stopped processing
(at offset 32606948 where log-end-offset is 33472402).
Broker itself is not reporting any active consumer in that consumer group and
the only info I was able to gather was from thread dump:
{code:java}
"mp_ads_publisher_pro_madstorage-web-corotos-prod-9db804ae-2a7a-431f-be09-392ab38cd8a2-StreamThread-33"
#113 prio=5 os_prio=0 tid=0x7fe07c6349b0 nid=0xf7a waiting on condition
[0x7fe0215d4000]
java.lang.Thread.State: WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0xfec6a2f8> (a
java.util.concurrent.CountDownLatch$Sync)
at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
at
java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836)
at
java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997)
at
java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304)
at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231)
at
org.apache.kafka.clients.producer.internals.TransactionalRequestResult.await(TransactionalRequestResult.java:50)
at
org.apache.kafka.clients.producer.KafkaProducer.initTransactions(KafkaProducer.java:554)
at
org.apache.kafka.streams.processor.internals.StreamTask.(StreamTask.java:151)
at
org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:404)
at
org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:365)
at
org.apache.kafka.streams.processor.internals.StreamThread$AbstractTaskCreator.createTasks(StreamThread.java:350)
at
org.apache.kafka.streams.processor.internals.TaskManager.addStreamTasks(TaskManager.java:137)
at
org.apache.kafka.streams.processor.internals.TaskManager.createTasks(TaskManager.java:88)
at
org.apache.kafka.streams.processor.internals.StreamThread$RebalanceListener.onPartitionsAssigned(StreamThread.java:259)
at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:264)
at
org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:367)
at
org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:316)
at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:295)
at
org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1146)
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:)
at
org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:851)
at
org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:808)
at
org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:774)
at
org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:744){code}
I tried restarting application once but the situation repeated. Thread read
some data, committed offset and stopped processing, leaving that thread in wait
state.
FYI: we have EOS enabled
was:
A kafka stream application thread stops processing without any feedback. The
topic has 24 partitions and I noticed that processing stopped only for some
partitions. I will describe what happened to partition:10. The application is
still running (now for about 8 hours) and that thread is hanging there and no
rebalancing that took place.
There is no error (we have a custom `Thread.UncaughtExceptionHandler` which was
not called). I noticed that after couple of minutes stream stopped processing
(at offset 32606948 where log-end-offset is 33472402).
Broker itself is not reporting any active consumer in that consumer group and
the only info I was able to gather was from thread dump:
{code:java}
"mp_ads_publisher_pro_madstorage-web-corotos-prod-9db804ae-2a7a-431f-be09-392ab38cd8a2-StreamThread-33"
#113 prio=5 os_prio=0 tid=0x7fe07c6349b0 nid=0xf7a waiting on condition
[0x7fe0215d4000]
java.lang.Thread.State: WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0xfec6a2f8> (a
java.util.concur
[jira] [Created] (KAFKA-7088) Kafka streams thread waits infinitely on transaction init
Lukasz Gluchowski created KAFKA-7088:
Summary: Kafka streams thread waits infinitely on transaction init
Key: KAFKA-7088
URL: https://issues.apache.org/jira/browse/KAFKA-7088
Project: Kafka
Issue Type: Bug
Components: streams
Affects Versions: 1.0.1
Environment: Linux 4.14.33-51.37.amzn1.x86_64 #1 SMP Thu May 3
20:07:43 UTC 2018
kafka-streams (client) 1.0.1
kafka broker 1.1.0
Java version:
OpenJDK Runtime Environment (build 1.8.0_171-b10)
OpenJDK 64-Bit Server VM (build 25.171-b10, mixed mode)
kakfa config overrides:
num.stream.threads: 6
session.timeout.ms: 1
request.timeout.ms: 11000
fetch.max.wait.ms: 500
max.poll.records: 1000
topic has 24 partitions
Reporter: Lukasz Gluchowski
A kafka stream application thread stops processing without any feedback. The
topic has 24 partitions and I noticed that processing stopped only for some
partitions. I will describe what happened to partition:10. The application is
still running (now for about 8 hours) and that thread is hanging there and no
rebalancing that took place.
There is no error (we have a custom `Thread.UncaughtExceptionHandler` which was
not called). I noticed that after couple of minutes stream stopped processing
(at offset 32606948 where log-end-offset is 33472402).
Broker itself is not reporting any active consumer in that consumer group and
the only info I was able to gather was from thread dump:
{code:java}
"mp_ads_publisher_pro_madstorage-web-corotos-prod-9db804ae-2a7a-431f-be09-392ab38cd8a2-StreamThread-33"
#113 prio=5 os_prio=0 tid=0x7fe07c6349b0 nid=0xf7a waiting on condition
[0x7fe0215d4000]
java.lang.Thread.State: WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0xfec6a2f8> (a
java.util.concurrent.CountDownLatch$Sync)
at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
at
java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836)
at
java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997)
at
java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304)
at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231)
at
org.apache.kafka.clients.producer.internals.TransactionalRequestResult.await(TransactionalRequestResult.java:50)
at
org.apache.kafka.clients.producer.KafkaProducer.initTransactions(KafkaProducer.java:554)
at
org.apache.kafka.streams.processor.internals.StreamTask.(StreamTask.java:151)
at
org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:404)
at
org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:365)
at
org.apache.kafka.streams.processor.internals.StreamThread$AbstractTaskCreator.createTasks(StreamThread.java:350)
at
org.apache.kafka.streams.processor.internals.TaskManager.addStreamTasks(TaskManager.java:137)
at
org.apache.kafka.streams.processor.internals.TaskManager.createTasks(TaskManager.java:88)
at
org.apache.kafka.streams.processor.internals.StreamThread$RebalanceListener.onPartitionsAssigned(StreamThread.java:259)
at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:264)
at
org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:367)
at
org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:316)
at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:295)
at
org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1146)
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:)
at
org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:851)
at
org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:808)
at
org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:774)
at
org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:744){code}
I tried restarting application once but the situation repeated. Thread read
some data, committed offset and stopped processing, leaving that thread in wait
state.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Created] (KAFKA-7087) Option in TopicCommand to report not preferred leaders
Viktor Somogyi created KAFKA-7087: - Summary: Option in TopicCommand to report not preferred leaders Key: KAFKA-7087 URL: https://issues.apache.org/jira/browse/KAFKA-7087 Project: Kafka Issue Type: Improvement Components: core Reporter: Viktor Somogyi Assignee: Viktor Somogyi Options in topic describe exists for reporting unavailable and lagging partitions but it is often an ask to report partitions where the active leader is not the preferred one. This jira adds this extra option to TopicCommand. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-7059) Offer new constructor on ProducerRecord
[
https://issues.apache.org/jira/browse/KAFKA-7059?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16519373#comment-16519373
]
ASF GitHub Bot commented on KAFKA-7059:
---
matzew closed pull request #5228: KAFKA-7059 Offer new constructor on
ProducerRecord
URL: https://github.com/apache/kafka/pull/5228
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/clients/src/main/java/org/apache/kafka/clients/producer/ProducerRecord.java
b/clients/src/main/java/org/apache/kafka/clients/producer/ProducerRecord.java
index 5d6bb3ff82e..8f34d14c8f1 100644
---
a/clients/src/main/java/org/apache/kafka/clients/producer/ProducerRecord.java
+++
b/clients/src/main/java/org/apache/kafka/clients/producer/ProducerRecord.java
@@ -107,7 +107,19 @@ public ProducerRecord(String topic, Integer partition,
Long timestamp, K key, V
public ProducerRecord(String topic, Integer partition, K key, V value,
Iterable headers) {
this(topic, partition, null, key, value, headers);
}
-
+
+/**
+ * Creates a record to be sent to a specified topic and partition
+ *
+ * @param topic The topic the record will be appended to
+ * @param key The key that will be included in the record
+ * @param value The record contents
+ * @param headers The headers that will be included in the record
+ */
+public ProducerRecord(String topic, K key, V value, Iterable
headers) {
+this(topic, null, null, key, value, headers);
+}
+
/**
* Creates a record to be sent to a specified topic and partition
*
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
> Offer new constructor on ProducerRecord
>
>
> Key: KAFKA-7059
> URL: https://issues.apache.org/jira/browse/KAFKA-7059
> Project: Kafka
> Issue Type: Improvement
>Reporter: Matthias Weßendorf
>Priority: Trivial
> Fix For: 2.0.1
>
>
> creating a ProducerRecord, with custom headers requires usage of a
> constructor with a slightly longer arguments list.
>
> It would be handy or more convenient if there was a ctor, like:
> {code}
> public ProducerRecord(String topic, K key, V value, Iterable headers)
> {code}
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (KAFKA-7086) Kafka server process dies after try deleting old log files under Windows 10
[
https://issues.apache.org/jira/browse/KAFKA-7086?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Cezary Wagner updated KAFKA-7086:
-
Labels: windows (was: )
> Kafka server process dies after try deleting old log files under Windows 10
> ---
>
> Key: KAFKA-7086
> URL: https://issues.apache.org/jira/browse/KAFKA-7086
> Project: Kafka
> Issue Type: Bug
> Components: core
>Affects Versions: 1.1.0
> Environment: Windows 10, Windows Server 2012 R2
>Reporter: Cezary Wagner
>Priority: Major
> Labels: windows
>
> Kafka after achieving log.retention.hours dies every time with error.
> {noformat}
> # Log Retention Policy
> #
> # The following configurations control the disposal of log segments. The
> policy can
> # be set to delete segments after a period of time, or after a given size has
> accumulated.
> # A segment will be deleted whenever *either* of these criteria are met.
> Deletion always happens
> # from the end of the log.
> # The minimum age of a log file to be eligible for deletion due to age
> log.retention.hours=168
> # A size-based retention policy for logs. Segments are pruned from the log
> unless the remaining
> # segments drop below log.retention.bytes. Functions independently of
> log.retention.hours.
> #log.retention.bytes=1073741824
> # The maximum size of a log segment file. When this size is reached a new log
> segment will be created.
> log.segment.bytes=1073741824
> # The interval at which log segments are checked to see if they can be
> deleted according
> # to the retention policies
> log.retention.check.interval.ms=30{noformat}
> Exception raised:
> {noformat}
> > C:\root\kafka_2.12-1.1.0\data\__consumer_offsets-3\.log.swap:
> > Proces nie mo┐e uzyskaŠ dostŕpu do pliku, poniewa┐ jest on u┐ywany przez
> > inny proces.
> at sun.nio.fs.WindowsException.translateToIOException(Unknown Source)
> at sun.nio.fs.WindowsException.rethrowAsIOException(Unknown Source)
> at sun.nio.fs.WindowsFileCopy.move(Unknown Source)
> at sun.nio.fs.WindowsFileSystemProvider.move(Unknown Source)
> at java.nio.file.Files.move(Unknown Source)
> at
> org.apache.kafka.common.utils.Utils.atomicMoveWithFallback(Utils.java:697)
> at
> org.apache.kafka.common.record.FileRecords.renameTo(FileRecords.java:212)
> at kafka.log.LogSegment.changeFileSuffixes(LogSegment.scala:415)
> at kafka.log.Log.replaceSegments(Log.scala:1644)
> at kafka.log.Cleaner.cleanSegments(LogCleaner.scala:535)
> at kafka.log.Cleaner.$anonfun$doClean$6(LogCleaner.scala:462)
> at kafka.log.Cleaner.$anonfun$doClean$6$adapted(LogCleaner.scala:461)
> at scala.collection.immutable.List.foreach(List.scala:389)
> at kafka.log.Cleaner.doClean(LogCleaner.scala:461)
> at kafka.log.Cleaner.clean(LogCleaner.scala:438)
> at
> kafka.log.LogCleaner$CleanerThread.cleanOrSleep(LogCleaner.scala:305)
> at kafka.log.LogCleaner$CleanerThread.doWork(LogCleaner.scala:291)
> at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:82)
> Suppressed: java.nio.file.FileSystemException:
> C:\root\kafka_2.12-1.1.0\data\__consumer_offsets-3\.log.cleaned
> ->
> C:\root\kafka_2.12-1.1.0\data\__consumer_offsets-3\.log.swap:
> Proces nie mo┐e uzyskaŠ dostŕpu do pliku, poniewa┐ jest on u┐ywany przez
> inny proces.
> at sun.nio.fs.WindowsException.translateToIOException(Unknown
> Source)
> at sun.nio.fs.WindowsException.rethrowAsIOException(Unknown
> Source)
> at sun.nio.fs.WindowsFileCopy.move(Unknown Source)
> at sun.nio.fs.WindowsFileSystemProvider.move(Unknown Source)
> at java.nio.file.Files.move(Unknown Source)
> at
> org.apache.kafka.common.utils.Utils.atomicMoveWithFallback(Utils.java:694)
> ... 12 more
> [2018-06-21 13:06:34,196] INFO [ReplicaManager broker=0] Stopping serving
> replicas in dir C:\root\kafka_2.12-1.1.0\data (kafka.server.ReplicaManager)
> [2018-06-21 13:06:34,209] INFO [ReplicaFetcherManager on broker 0] Removed
> fetcher for partitions
> __consumer_offsets-22,test-14,__consumer_offsets-30,test-6,__consumer_offsets-8,__consumer_offsets-21,test-17,__consumer_offsets-4,INTEGRATION_TESTS_DEBUG.monitoring_events-0,test-20,__consumer_offsets-27,__consumer_offsets-7,__consumer_offsets-9,test-29,__consumer_offsets-46,test-23,test-24,test-11,test-10,__consumer_offsets-25,__consumer_offsets-35,__consumer_offsets-41,__consumer_offsets-33,__consumer_offsets-23,__consumer_offsets-49,test-28,__consumer_offsets-47,test-19
[jira] [Created] (KAFKA-7086) Kafka server process dies after try deleting old log files under Windows 10
Cezary Wagner created KAFKA-7086:
Summary: Kafka server process dies after try deleting old log
files under Windows 10
Key: KAFKA-7086
URL: https://issues.apache.org/jira/browse/KAFKA-7086
Project: Kafka
Issue Type: Bug
Components: core
Affects Versions: 1.1.0
Environment: Windows 10, Windows Server 2012 R2
Reporter: Cezary Wagner
Kafka after achieving log.retention.hours dies every time with error.
{noformat}
# Log Retention Policy #
# The following configurations control the disposal of log segments. The policy
can
# be set to delete segments after a period of time, or after a given size has
accumulated.
# A segment will be deleted whenever *either* of these criteria are met.
Deletion always happens
# from the end of the log.
# The minimum age of a log file to be eligible for deletion due to age
log.retention.hours=168
# A size-based retention policy for logs. Segments are pruned from the log
unless the remaining
# segments drop below log.retention.bytes. Functions independently of
log.retention.hours.
#log.retention.bytes=1073741824
# The maximum size of a log segment file. When this size is reached a new log
segment will be created.
log.segment.bytes=1073741824
# The interval at which log segments are checked to see if they can be deleted
according
# to the retention policies
log.retention.check.interval.ms=30{noformat}
Exception raised:
{noformat}
> C:\root\kafka_2.12-1.1.0\data\__consumer_offsets-3\.log.swap:
> Proces nie mo┐e uzyskaŠ dostŕpu do pliku, poniewa┐ jest on u┐ywany przez
> inny proces.
at sun.nio.fs.WindowsException.translateToIOException(Unknown Source)
at sun.nio.fs.WindowsException.rethrowAsIOException(Unknown Source)
at sun.nio.fs.WindowsFileCopy.move(Unknown Source)
at sun.nio.fs.WindowsFileSystemProvider.move(Unknown Source)
at java.nio.file.Files.move(Unknown Source)
at
org.apache.kafka.common.utils.Utils.atomicMoveWithFallback(Utils.java:697)
at
org.apache.kafka.common.record.FileRecords.renameTo(FileRecords.java:212)
at kafka.log.LogSegment.changeFileSuffixes(LogSegment.scala:415)
at kafka.log.Log.replaceSegments(Log.scala:1644)
at kafka.log.Cleaner.cleanSegments(LogCleaner.scala:535)
at kafka.log.Cleaner.$anonfun$doClean$6(LogCleaner.scala:462)
at kafka.log.Cleaner.$anonfun$doClean$6$adapted(LogCleaner.scala:461)
at scala.collection.immutable.List.foreach(List.scala:389)
at kafka.log.Cleaner.doClean(LogCleaner.scala:461)
at kafka.log.Cleaner.clean(LogCleaner.scala:438)
at kafka.log.LogCleaner$CleanerThread.cleanOrSleep(LogCleaner.scala:305)
at kafka.log.LogCleaner$CleanerThread.doWork(LogCleaner.scala:291)
at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:82)
Suppressed: java.nio.file.FileSystemException:
C:\root\kafka_2.12-1.1.0\data\__consumer_offsets-3\.log.cleaned
->
C:\root\kafka_2.12-1.1.0\data\__consumer_offsets-3\.log.swap:
Proces nie mo┐e uzyskaŠ dostŕpu do pliku, poniewa┐ jest on u┐ywany przez inny
proces.
at sun.nio.fs.WindowsException.translateToIOException(Unknown
Source)
at sun.nio.fs.WindowsException.rethrowAsIOException(Unknown
Source)
at sun.nio.fs.WindowsFileCopy.move(Unknown Source)
at sun.nio.fs.WindowsFileSystemProvider.move(Unknown Source)
at java.nio.file.Files.move(Unknown Source)
at
org.apache.kafka.common.utils.Utils.atomicMoveWithFallback(Utils.java:694)
... 12 more
[2018-06-21 13:06:34,196] INFO [ReplicaManager broker=0] Stopping serving
replicas in dir C:\root\kafka_2.12-1.1.0\data (kafka.server.ReplicaManager)
[2018-06-21 13:06:34,209] INFO [ReplicaFetcherManager on broker 0] Removed
fetcher for partitions
__consumer_offsets-22,test-14,__consumer_offsets-30,test-6,__consumer_offsets-8,__consumer_offsets-21,test-17,__consumer_offsets-4,INTEGRATION_TESTS_DEBUG.monitoring_events-0,test-20,__consumer_offsets-27,__consumer_offsets-7,__consumer_offsets-9,test-29,__consumer_offsets-46,test-23,test-24,test-11,test-10,__consumer_offsets-25,__consumer_offsets-35,__consumer_offsets-41,__consumer_offsets-33,__consumer_offsets-23,__consumer_offsets-49,test-28,__consumer_offsets-47,test-19,__consumer_offsets-16,test-0,__consumer_offsets-28,test-7,__consumer_offsets-31,__consumer_offsets-36,__consumer_offsets-42,__consumer_offsets-3,test-18,__consumer_offsets-18,test-22,test-25,__consumer_offsets-37,test-5,__consumer_offsets-15,__consumer_offsets-24,__consumer_offsets-38,__consumer_offsets-17,test-8,__consumer_offsets-48,test-1,__consumer_offsets-19,test-26,__consumer_offsets-11,__consumer_offsets-13,
[jira] [Created] (KAFKA-7085) Producer hangs on TimeoutException: Failed to update metadata after 60000 ms.
Martin Vysny created KAFKA-7085: --- Summary: Producer hangs on TimeoutException: Failed to update metadata after 6 ms. Key: KAFKA-7085 URL: https://issues.apache.org/jira/browse/KAFKA-7085 Project: Kafka Issue Type: Bug Components: clients Affects Versions: 1.1.0 Reporter: Martin Vysny I start Kafka in Docker: docker run --rm -d -p 2181:2181 -p 9092:9092 --env ADVERTISED_HOST="127.0.0.1" spotify/kafka Then, I connect KafkaProducer to that Kafka (127.0.0.1:9092) and immediately I call producer.send() . The send() method always blocks for 60s and then fails and callback receives TimeoutException. The producer is therefore unable to send any message which renders it useless. A workaround is to sleep for 1-2 seconds after the Producer is constructed and before first send() is invoked - that apparently gives enough time to Kafka to synchronize whatever it needs and no longer block endlessly in send() -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-7084) NewTopicBuilder#config should accept Map rather than Map
[ https://issues.apache.org/jira/browse/KAFKA-7084?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16519146#comment-16519146 ] ASF GitHub Bot commented on KAFKA-7084: --- chia7712 opened a new pull request #5261: KAFKA-7084 NewTopicBuilder#config should accept Map r… URL: https://github.com/apache/kafka/pull/5261 ### 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 > NewTopicBuilder#config should accept Map rather than > Map > > > Key: KAFKA-7084 > URL: https://issues.apache.org/jira/browse/KAFKA-7084 > Project: Kafka > Issue Type: Improvement >Reporter: Chia-Ping Tsai >Assignee: Chia-Ping Tsai >Priority: Major > > The field "config" in NewTopic is Map[String, String] but > NewTopicBuilder#config accept the Map[String, Object] and then call > Object#toString to convert the Map[String, Object] to Map[String, String]. > That is weird since users have to trace the source code to understand how > kafka generate the Map[String, String]. > we should deprecate NewTopicBuilder#config(Map[String, Object]) and add an > alternative method NewTopicBuilder#config"s"(Map[String, String]) -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (KAFKA-7084) NewTopicBuilder#config should accept Map rather than Map
Chia-Ping Tsai created KAFKA-7084: - Summary: NewTopicBuilder#config should accept Map rather than Map Key: KAFKA-7084 URL: https://issues.apache.org/jira/browse/KAFKA-7084 Project: Kafka Issue Type: Improvement Reporter: Chia-Ping Tsai Assignee: Chia-Ping Tsai The field "config" in NewTopic is Map[String, String] but NewTopicBuilder#config accept the Map[String, Object] and then call Object#toString to convert the Map[String, Object] to Map[String, String]. That is weird since users have to trace the source code to understand how kafka generate the Map[String, String]. we should deprecate NewTopicBuilder#config(Map[String, Object]) and add an alternative method NewTopicBuilder#config"s"(Map[String, String]) -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-7083) maxTickMessages in ConsumerGroup option
[ https://issues.apache.org/jira/browse/KAFKA-7083?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16518999#comment-16518999 ] Manikumar commented on KAFKA-7083: -- looks like you are using Node.js Kafka client. Can you post this query on Node.js client mailing list/github issue list > maxTickMessages in ConsumerGroup option > --- > > Key: KAFKA-7083 > URL: https://issues.apache.org/jira/browse/KAFKA-7083 > Project: Kafka > Issue Type: Bug >Reporter: Rahul >Priority: Major > > Hello, > We are using Kafka v2.11.1.0.0. We have only one partition for our topic. > While consuming a record from the topic, I am setting maxTickMessages = 1 in > Kafka consumer group. It returns me 2 records. I am not getting why it is > giving me 1 extra record of the mentioned size. Whenever I increase the > number in maxTickMessages, it gives me one extra record. > Can someone please suggest me a solution to this issue? > Thanks & Regards, > Rahul Singh -- This message was sent by Atlassian JIRA (v7.6.3#76005)
