[jira] [Resolved] (KAFKA-8624) 这边是服务端的kafka，版本是 kafka_2.11-1.0.0.2，当客户端向主题发送消息的时候抛出如下异常，客户端跟服务端的版本不一致吗？
[
https://issues.apache.org/jira/browse/KAFKA-8624?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
CHARELS resolved KAFKA-8624.
Resolution: Fixed
Fix Version/s: 1.1.0
可能是produer升级了client版本，导致kafka的服务端消息形成了header不一致了，然后0.10.*版本的consumer无法消费，换成1.1.0版本可正常消费。
> 这边是服务端的kafka，版本是 kafka_2.11-1.0.0.2，当客户端向主题发送消息的时候抛出如下异常，客户端跟服务端的版本不一致吗？
>
>
> Key: KAFKA-8624
> URL: https://issues.apache.org/jira/browse/KAFKA-8624
> Project: Kafka
> Issue Type: Bug
> Components: log
>Affects Versions: 1.0.0
>Reporter: CHARELS
>Priority: Major
> Labels: pull-request-available
> Fix For: 1.1.0
>
> Attachments: image-2019-07-04-00-18-15-781.png,
> image-2019-07-04-14-49-47-802.png, image-2019-07-04-14-56-24-216.png, 李志涛.jpg
>
> Original Estimate: 12h
> Remaining Estimate: 12h
>
> ERROR [KafkaApi-1004] Error when handling request
> \{replica_id=-1,max_wait_time=1,min_bytes=0,topics=[{topic=ETE,partitions=[{partition=3,fetch_offset=119224671,max_bytes=1048576}]}]}
> (kafka.server.KafkaApis)
> java.lang.IllegalArgumentException: Magic v0 does not support record headers
> at
> org.apache.kafka.common.record.MemoryRecordsBuilder.appendWithOffset(MemoryRecordsBuilder.java:403)
> at
> org.apache.kafka.common.record.MemoryRecordsBuilder.append(MemoryRecordsBuilder.java:586)
> at
> org.apache.kafka.common.record.AbstractRecords.convertRecordBatch(AbstractRecords.java:134)
> at
> org.apache.kafka.common.record.AbstractRecords.downConvert(AbstractRecords.java:109)
> at
> org.apache.kafka.common.record.FileRecords.downConvert(FileRecords.java:253)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$convertedPartitionData$1$1$$anonfun$apply$4.apply(KafkaApis.scala:519)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$convertedPartitionData$1$1$$anonfun$apply$4.apply(KafkaApis.scala:517)
> at scala.Option.map(Option.scala:146)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$convertedPartitionData$1$1.apply(KafkaApis.scala:517)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$convertedPartitionData$1$1.apply(KafkaApis.scala:507)
> at scala.Option.flatMap(Option.scala:171)
> at
> kafka.server.KafkaApis.kafka$server$KafkaApis$$convertedPartitionData$1(KafkaApis.scala:507)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$createResponse$2$1.apply(KafkaApis.scala:555)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$createResponse$2$1.apply(KafkaApis.scala:554)
> 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.server.KafkaApis.kafka$server$KafkaApis$$createResponse$2(KafkaApis.scala:554)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$fetchResponseCallback$1$1.apply(KafkaApis.scala:568)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$fetchResponseCallback$1$1.apply(KafkaApis.scala:568)
> at
> kafka.server.KafkaApis$$anonfun$sendResponseMaybeThrottle$1.apply$mcVI$sp(KafkaApis.scala:2033)
> at
> kafka.server.ClientRequestQuotaManager.maybeRecordAndThrottle(ClientRequestQuotaManager.scala:54)
> at kafka.server.KafkaApis.sendResponseMaybeThrottle(KafkaApis.scala:2032)
> at
> kafka.server.KafkaApis.kafka$server$KafkaApis$$fetchResponseCallback$1(KafkaApis.scala:568)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$processResponseCallback$1$1.apply$mcVI$sp(KafkaApis.scala:587)
> at
> kafka.server.ClientQuotaManager.maybeRecordAndThrottle(ClientQuotaManager.scala:176)
> at
> kafka.server.KafkaApis.kafka$server$KafkaApis$$processResponseCallback$1(KafkaApis.scala:586)
> at
> kafka.server.KafkaApis$$anonfun$handleFetchRequest$3.apply(KafkaApis.scala:603)
> at
> kafka.server.KafkaApis$$anonfun$handleFetchRequest$3.apply(KafkaApis.scala:603)
> at kafka.server.ReplicaManager.fetchMessages(ReplicaManager.scala:820)
> at kafka.server.KafkaApis.handleFetchRequest(KafkaApis.scala:595)
> at kafka.server.KafkaApis.handle(KafkaApis.scala:99)
> at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:65)
> at java.lang.Thread.run(Thread.java:748)
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-6582) Partitions get underreplicated, with a single ISR, and doesn't recover. Other brokers do not take over and we need to manually restart the broker.
[
https://issues.apache.org/jira/browse/KAFKA-6582?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16878859#comment-16878859
]
Sebastian Schmitz commented on KAFKA-6582:
--
We are running 2.1.1 in production (and four other environments) since March
without this issue showing again.
> Partitions get underreplicated, with a single ISR, and doesn't recover. Other
> brokers do not take over and we need to manually restart the broker.
> --
>
> Key: KAFKA-6582
> URL: https://issues.apache.org/jira/browse/KAFKA-6582
> Project: Kafka
> Issue Type: Bug
> Components: network
>Affects Versions: 1.0.0
> Environment: Ubuntu 16.04
> Linux kafka04 4.4.0-109-generic #132-Ubuntu SMP Tue Jan 9 19:52:39 UTC 2018
> x86_64 x86_64 x86_64 GNU/Linux
> java version "9.0.1"
> Java(TM) SE Runtime Environment (build 9.0.1+11)
> Java HotSpot(TM) 64-Bit Server VM (build 9.0.1+11, mixed mode)
> but also tried with the latest JVM 8 before with the same result.
>Reporter: Jurriaan Pruis
>Priority: Major
> Attachments: Screenshot 2019-01-18 at 13.08.17.png, Screenshot
> 2019-01-18 at 13.16.59.png
>
>
> Partitions get underreplicated, with a single ISR, and doesn't recover. Other
> brokers do not take over and we need to manually restart the 'single ISR'
> broker (if you describe the partitions of replicated topic it is clear that
> some partitions are only in sync on this broker).
> This bug resembles KAFKA-4477 a lot, but since that issue is marked as
> resolved this is probably something else but similar.
> We have the same issue (or at least it looks pretty similar) on Kafka 1.0.
> Since upgrading to Kafka 1.0 in November 2017 we've had these issues (we've
> upgraded from Kafka 0.10.2.1).
> This happens almost every 24-48 hours on a random broker. This is why we
> currently have a cronjob which restarts every broker every 24 hours.
> During this issue the ISR shows the following server log:
> {code:java}
> [2018-02-20 12:02:08,342] WARN Attempting to send response via channel for
> which there is no open connection, connection id
> 10.132.0.32:9092-10.14.148.20:56352-96708 (kafka.network.Processor)
> [2018-02-20 12:02:08,364] WARN Attempting to send response via channel for
> which there is no open connection, connection id
> 10.132.0.32:9092-10.14.150.25:54412-96715 (kafka.network.Processor)
> [2018-02-20 12:02:08,349] WARN Attempting to send response via channel for
> which there is no open connection, connection id
> 10.132.0.32:9092-10.14.149.18:35182-96705 (kafka.network.Processor)
> [2018-02-20 12:02:08,379] WARN Attempting to send response via channel for
> which there is no open connection, connection id
> 10.132.0.32:9092-10.14.150.25:54456-96717 (kafka.network.Processor)
> [2018-02-20 12:02:08,448] WARN Attempting to send response via channel for
> which there is no open connection, connection id
> 10.132.0.32:9092-10.14.159.20:36388-96720 (kafka.network.Processor)
> [2018-02-20 12:02:08,683] WARN Attempting to send response via channel for
> which there is no open connection, connection id
> 10.132.0.32:9092-10.14.157.110:41922-96740 (kafka.network.Processor)
> {code}
> Also on the ISR broker, the controller log shows this:
> {code:java}
> [2018-02-20 12:02:14,927] INFO [Controller-3-to-broker-3-send-thread]:
> Controller 3 connected to 10.132.0.32:9092 (id: 3 rack: null) for sending
> state change requests (kafka.controller.RequestSendThread)
> [2018-02-20 12:02:14,927] INFO [Controller-3-to-broker-0-send-thread]:
> Controller 3 connected to 10.132.0.10:9092 (id: 0 rack: null) for sending
> state change requests (kafka.controller.RequestSendThread)
> [2018-02-20 12:02:14,928] INFO [Controller-3-to-broker-1-send-thread]:
> Controller 3 connected to 10.132.0.12:9092 (id: 1 rack: null) for sending
> state change requests (kafka.controller.RequestSendThread){code}
> And the non-ISR brokers show these kind of errors:
>
> {code:java}
> 2018-02-20 12:02:29,204] WARN [ReplicaFetcher replicaId=1, leaderId=3,
> fetcherId=0] Error in fetch to broker 3, request (type=FetchRequest,
> replicaId=1, maxWait=500, minBytes=1, maxBytes=10485760,
> fetchData={..}, isolationLevel=READ_UNCOMMITTED)
> (kafka.server.ReplicaFetcherThread)
> java.io.IOException: Connection to 3 was disconnected before the response was
> read
> at
> org.apache.kafka.clients.NetworkClientUtils.sendAndReceive(NetworkClientUtils.java:95)
> at
> kafka.server.ReplicaFetcherBlockingSend.sendRequest(ReplicaFetcherBlockingSend.scala:96)
> at kafka.server.ReplicaFetcherThread.fetch(ReplicaFetcherThread.scala:205)
> at kafka.server.ReplicaFetcherThread.fetch(ReplicaFetcherThread
[jira] [Commented] (KAFKA-6582) Partitions get underreplicated, with a single ISR, and doesn't recover. Other brokers do not take over and we need to manually restart the broker.
[
https://issues.apache.org/jira/browse/KAFKA-6582?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16878797#comment-16878797
]
qkafew commented on KAFKA-6582:
---
Hello，
I encountered the same problem in version 1.0.0.
How to fix it in 2.1.1?
Thanks.
> Partitions get underreplicated, with a single ISR, and doesn't recover. Other
> brokers do not take over and we need to manually restart the broker.
> --
>
> Key: KAFKA-6582
> URL: https://issues.apache.org/jira/browse/KAFKA-6582
> Project: Kafka
> Issue Type: Bug
> Components: network
>Affects Versions: 1.0.0
> Environment: Ubuntu 16.04
> Linux kafka04 4.4.0-109-generic #132-Ubuntu SMP Tue Jan 9 19:52:39 UTC 2018
> x86_64 x86_64 x86_64 GNU/Linux
> java version "9.0.1"
> Java(TM) SE Runtime Environment (build 9.0.1+11)
> Java HotSpot(TM) 64-Bit Server VM (build 9.0.1+11, mixed mode)
> but also tried with the latest JVM 8 before with the same result.
>Reporter: Jurriaan Pruis
>Priority: Major
> Attachments: Screenshot 2019-01-18 at 13.08.17.png, Screenshot
> 2019-01-18 at 13.16.59.png
>
>
> Partitions get underreplicated, with a single ISR, and doesn't recover. Other
> brokers do not take over and we need to manually restart the 'single ISR'
> broker (if you describe the partitions of replicated topic it is clear that
> some partitions are only in sync on this broker).
> This bug resembles KAFKA-4477 a lot, but since that issue is marked as
> resolved this is probably something else but similar.
> We have the same issue (or at least it looks pretty similar) on Kafka 1.0.
> Since upgrading to Kafka 1.0 in November 2017 we've had these issues (we've
> upgraded from Kafka 0.10.2.1).
> This happens almost every 24-48 hours on a random broker. This is why we
> currently have a cronjob which restarts every broker every 24 hours.
> During this issue the ISR shows the following server log:
> {code:java}
> [2018-02-20 12:02:08,342] WARN Attempting to send response via channel for
> which there is no open connection, connection id
> 10.132.0.32:9092-10.14.148.20:56352-96708 (kafka.network.Processor)
> [2018-02-20 12:02:08,364] WARN Attempting to send response via channel for
> which there is no open connection, connection id
> 10.132.0.32:9092-10.14.150.25:54412-96715 (kafka.network.Processor)
> [2018-02-20 12:02:08,349] WARN Attempting to send response via channel for
> which there is no open connection, connection id
> 10.132.0.32:9092-10.14.149.18:35182-96705 (kafka.network.Processor)
> [2018-02-20 12:02:08,379] WARN Attempting to send response via channel for
> which there is no open connection, connection id
> 10.132.0.32:9092-10.14.150.25:54456-96717 (kafka.network.Processor)
> [2018-02-20 12:02:08,448] WARN Attempting to send response via channel for
> which there is no open connection, connection id
> 10.132.0.32:9092-10.14.159.20:36388-96720 (kafka.network.Processor)
> [2018-02-20 12:02:08,683] WARN Attempting to send response via channel for
> which there is no open connection, connection id
> 10.132.0.32:9092-10.14.157.110:41922-96740 (kafka.network.Processor)
> {code}
> Also on the ISR broker, the controller log shows this:
> {code:java}
> [2018-02-20 12:02:14,927] INFO [Controller-3-to-broker-3-send-thread]:
> Controller 3 connected to 10.132.0.32:9092 (id: 3 rack: null) for sending
> state change requests (kafka.controller.RequestSendThread)
> [2018-02-20 12:02:14,927] INFO [Controller-3-to-broker-0-send-thread]:
> Controller 3 connected to 10.132.0.10:9092 (id: 0 rack: null) for sending
> state change requests (kafka.controller.RequestSendThread)
> [2018-02-20 12:02:14,928] INFO [Controller-3-to-broker-1-send-thread]:
> Controller 3 connected to 10.132.0.12:9092 (id: 1 rack: null) for sending
> state change requests (kafka.controller.RequestSendThread){code}
> And the non-ISR brokers show these kind of errors:
>
> {code:java}
> 2018-02-20 12:02:29,204] WARN [ReplicaFetcher replicaId=1, leaderId=3,
> fetcherId=0] Error in fetch to broker 3, request (type=FetchRequest,
> replicaId=1, maxWait=500, minBytes=1, maxBytes=10485760,
> fetchData={..}, isolationLevel=READ_UNCOMMITTED)
> (kafka.server.ReplicaFetcherThread)
> java.io.IOException: Connection to 3 was disconnected before the response was
> read
> at
> org.apache.kafka.clients.NetworkClientUtils.sendAndReceive(NetworkClientUtils.java:95)
> at
> kafka.server.ReplicaFetcherBlockingSend.sendRequest(ReplicaFetcherBlockingSend.scala:96)
> at kafka.server.ReplicaFetcherThread.fetch(ReplicaFetcherThread.scala:205)
> at kafka.server.ReplicaFetcherThread.fetch(ReplicaFetcherThread.scala:41)
> at
> kafka.server.AbstractFe
[jira] [Comment Edited] (KAFKA-8629) Kafka Streams Apps to support small native images through GraalVM
[ https://issues.apache.org/jira/browse/KAFKA-8629?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16878376#comment-16878376 ] Andy Muir edited comment on KAFKA-8629 at 7/4/19 4:42 PM: -- Kafka fork with modified code: [https://github.com/muirandy/kafka/tree/graal-native-image-support] Shows whats needed to make Kafka Streams app work with GraalVM native image. Note: More changes may be required, and as stated in the description above, this code is not ready for a PR. NB: This is my original work and I license the work to the project under the project's open source license. was (Author: muirandy): Kafka fork with modified code: [https://github.com/muirandy/kafka/tree/graal-native-image-support] Shows whats needed to make Kafka Streams app work with GraalVM native image. Note: More changes may be required, and as stated in the description above, this code is not ready for a PR. > Kafka Streams Apps to support small native images through GraalVM > - > > Key: KAFKA-8629 > URL: https://issues.apache.org/jira/browse/KAFKA-8629 > Project: Kafka > Issue Type: Improvement > Components: streams >Affects Versions: 2.3.0 > Environment: OSX > Linux on Docker >Reporter: Andy Muir >Priority: Minor > > I'm investigating using [GraalVM|http://example.com/] to help with reducing > docker image size and required resources for a simple Kafka Streams > microservice. To this end, I'm looking at running a microservice which: > 1) consumes from a Kafka topic (XML) > 2) Transforms into JSON > 3) Produces to a new Kafka topic. > The Kafka Streams app running in the JVM works fine. > When I attempt to build it to a GraalVM native image (binary executable which > does not require the JVM, hence smaller image size and less resources), I > encountered a few > [incompatibilities|https://github.com/oracle/graal/blob/master/substratevm/LIMITATIONS.md] > with the source code in Kafka. > I've implemented a workaround for each of these in a fork (link to follow) to > help establish if it is feasible. I don't intend (at this stage) for the > changes to be applied to the broker - I'm only after Kafka Streams for now. > I'm not sure whether it'd be a good idea for the broker itself to run as a > native image! > There were 2 issues getting the native image with kafka streams: > 1) Some Reflection use cases using MethodHandle > 2) Anything JMX > To work around these issues, I have: > 1) Replaced use of MethodHandle with alternatives > 2) Commented out the JMX code in a few places > While the first may be sustainable, I'd expect that the 2nd option should be > put behind a configuration switch to allow the existing code to be used by > default and turning off JMX if configured. > *I haven't created a PR for now, as I'd like feedback to decide if it is > going to be feasible to carry this forwards.* -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Resolved] (KAFKA-8443) Allow broker to select a preferred read replica for consumer
[ https://issues.apache.org/jira/browse/KAFKA-8443?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jason Gustafson resolved KAFKA-8443. Resolution: Fixed Fix Version/s: 2.4.0 > Allow broker to select a preferred read replica for consumer > > > Key: KAFKA-8443 > URL: https://issues.apache.org/jira/browse/KAFKA-8443 > Project: Kafka > Issue Type: Task >Reporter: David Arthur >Assignee: David Arthur >Priority: Minor > Fix For: 2.4.0 > > > This tracks the broker-side implementation of > [KIP-392|https://cwiki.apache.org/confluence/display/KAFKA/KIP-392%3A+Allow+consumers+to+fetch+from+closest+replica] > (fetch from follower) -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-8443) Allow broker to select a preferred read replica for consumer
[ https://issues.apache.org/jira/browse/KAFKA-8443?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16878734#comment-16878734 ] ASF GitHub Bot commented on KAFKA-8443: --- hachikuji commented on pull request #6832: KAFKA-8443 Broker support for fetch from followers URL: https://github.com/apache/kafka/pull/6832 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Allow broker to select a preferred read replica for consumer > > > Key: KAFKA-8443 > URL: https://issues.apache.org/jira/browse/KAFKA-8443 > Project: Kafka > Issue Type: Task >Reporter: David Arthur >Assignee: David Arthur >Priority: Minor > > This tracks the broker-side implementation of > [KIP-392|https://cwiki.apache.org/confluence/display/KAFKA/KIP-392%3A+Allow+consumers+to+fetch+from+closest+replica] > (fetch from follower) -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-6654) Customize SSLContext creation
[ https://issues.apache.org/jira/browse/KAFKA-6654?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16878731#comment-16878731 ] Igor Soarez commented on KAFKA-6654: What's the state of this? Would you mind if I pick this up? [~clement_pelle...@ibi.com] > Customize SSLContext creation > - > > Key: KAFKA-6654 > URL: https://issues.apache.org/jira/browse/KAFKA-6654 > Project: Kafka > Issue Type: Improvement > Components: config >Affects Versions: 1.0.0 >Reporter: Robert Wruck >Priority: Major > > Currently, loading of SSL keystore and truststore always uses a > FileInputStream (SslFactory.SecurityStore) and cannot be changed to load > keystores from other locations such as the classpath, raw byte arrays etc. > Furthermore, passwords for the key stores have to be provided as plaintext > configuration properties. > Delegating the creation of an SSLContext to a customizable implementation > might solve some more issues such as KAFKA-5519, KAFKA-4933, KAFKA-4294, > KAFKA-2629 by enabling Kafka users to implement their own. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-5998) /.checkpoint.tmp Not found exception
[
https://issues.apache.org/jira/browse/KAFKA-5998?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16878674#comment-16878674
]
Di Campo commented on KAFKA-5998:
-
My topology contains stateless subtopologies, so it may be that case.
One question:
> The "trigger" for the cleanup is the absence of any traffic on a topic
> partition which will allow the directory to be deleted
Can you elaborate on this trigger? What will be cleaned in this cleanup
process? How long does it wait by default for traffic to determine absence?
> /.checkpoint.tmp Not found exception
>
>
> Key: KAFKA-5998
> URL: https://issues.apache.org/jira/browse/KAFKA-5998
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 0.11.0.0, 0.11.0.1, 2.1.1
>Reporter: Yogesh BG
>Assignee: Bill Bejeck
>Priority: Critical
> Attachments: 5998.v1.txt, 5998.v2.txt, Kafka5998.zip, Topology.txt,
> exc.txt, props.txt, streams.txt
>
>
> I have one kafka broker and one kafka stream running... I am running its
> since two days under load of around 2500 msgs per second.. On third day am
> getting below exception for some of the partitions, I have 16 partitions only
> 0_0 and 0_1 gives this error
> {{09:43:25.955 [ks_0_inst-StreamThread-6] WARN
> o.a.k.s.p.i.ProcessorStateManager - Failed to write checkpoint file to
> /data/kstreams/rtp-kafkastreams/0_1/.checkpoint:
> java.io.FileNotFoundException:
> /data/kstreams/rtp-kafkastreams/0_1/.checkpoint.tmp (No such file or
> directory)
> at java.io.FileOutputStream.open(Native Method) ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:221)
> ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:171)
> ~[na:1.7.0_111]
> at
> org.apache.kafka.streams.state.internals.OffsetCheckpoint.write(OffsetCheckpoint.java:73)
> ~[rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.ProcessorStateManager.checkpoint(ProcessorStateManager.java:324)
> ~[rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask$1.run(StreamTask.java:267)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:201)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:260)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:254)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.AssignedTasks$1.apply(AssignedTasks.java:322)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.AssignedTasks.applyToRunningTasks(AssignedTasks.java:415)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.AssignedTasks.commit(AssignedTasks.java:314)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.commitAll(StreamThread.java:700)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:683)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:523)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:480)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:457)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> 09:43:25.974 [ks_0_inst-StreamThread-15] WARN
> o.a.k.s.p.i.ProcessorStateManager - Failed to write checkpoint file to
> /data/kstreams/rtp-kafkastreams/0_0/.checkpoint:
> java.io.FileNotFoundException:
> /data/kstreams/rtp-kafkastreams/0_0/.checkpoint.tmp (No such file or
> directory)
> at java.io.FileOutputStream.open(Native Method) ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:221)
> ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:171)
> ~[na:1.7.0_111]
> at
> org.apache.kafka.streams.state.internals.OffsetCheckpoint.write(OffsetC
[jira] [Comment Edited] (KAFKA-5998) /.checkpoint.tmp Not found exception
[
https://issues.apache.org/jira/browse/KAFKA-5998?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16878674#comment-16878674
]
Di Campo edited comment on KAFKA-5998 at 7/4/19 1:55 PM:
-
My topology contains stateless subtopologies as well as stateful, so it may be
that case.
One question:
> The "trigger" for the cleanup is the absence of any traffic on a topic
> partition which will allow the directory to be deleted
Can you elaborate on this trigger? What will be cleaned in this cleanup
process? How long does it wait by default for traffic to determine absence?
was (Author: xmar):
My topology contains stateless subtopologies, so it may be that case.
One question:
> The "trigger" for the cleanup is the absence of any traffic on a topic
> partition which will allow the directory to be deleted
Can you elaborate on this trigger? What will be cleaned in this cleanup
process? How long does it wait by default for traffic to determine absence?
> /.checkpoint.tmp Not found exception
>
>
> Key: KAFKA-5998
> URL: https://issues.apache.org/jira/browse/KAFKA-5998
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 0.11.0.0, 0.11.0.1, 2.1.1
>Reporter: Yogesh BG
>Assignee: Bill Bejeck
>Priority: Critical
> Attachments: 5998.v1.txt, 5998.v2.txt, Kafka5998.zip, Topology.txt,
> exc.txt, props.txt, streams.txt
>
>
> I have one kafka broker and one kafka stream running... I am running its
> since two days under load of around 2500 msgs per second.. On third day am
> getting below exception for some of the partitions, I have 16 partitions only
> 0_0 and 0_1 gives this error
> {{09:43:25.955 [ks_0_inst-StreamThread-6] WARN
> o.a.k.s.p.i.ProcessorStateManager - Failed to write checkpoint file to
> /data/kstreams/rtp-kafkastreams/0_1/.checkpoint:
> java.io.FileNotFoundException:
> /data/kstreams/rtp-kafkastreams/0_1/.checkpoint.tmp (No such file or
> directory)
> at java.io.FileOutputStream.open(Native Method) ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:221)
> ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:171)
> ~[na:1.7.0_111]
> at
> org.apache.kafka.streams.state.internals.OffsetCheckpoint.write(OffsetCheckpoint.java:73)
> ~[rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.ProcessorStateManager.checkpoint(ProcessorStateManager.java:324)
> ~[rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask$1.run(StreamTask.java:267)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:201)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:260)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:254)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.AssignedTasks$1.apply(AssignedTasks.java:322)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.AssignedTasks.applyToRunningTasks(AssignedTasks.java:415)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.AssignedTasks.commit(AssignedTasks.java:314)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.commitAll(StreamThread.java:700)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:683)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:523)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:480)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:457)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> 09:43:25.974 [ks_0_inst-StreamThread-15] WARN
> o.a.k.s.p.i.ProcessorStateManager - Failed to write checkpoint file to
> /data/kstreams/rtp-kafkastreams/0_0/.ch
[jira] [Updated] (KAFKA-8630) Unit testing a streams processor with a WindowStore throws a ClassCastException
[
https://issues.apache.org/jira/browse/KAFKA-8630?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Justin Fetherolf updated KAFKA-8630:
Description:
I was attempting to write a unit test for a class implementing the
{{Processor}} interface that contained a {{WindowStore}}, but running the test
fails with a {{ClassCastException}} coming out of {{InMemoryWindowStore.init}}
attempting to cast {{MockProcessorContext}} to {{InternalProcessorContext}}.
Minimal code to reproduce:
{code:java}
package com.cantgetthistowork;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.state.WindowStore;
public class InMemWindowProcessor implements Processor {
private ProcessorContext context;
private WindowStore windowStore;
@Override
public void init(ProcessorContext context) {
this.context = context;
windowStore = (WindowStore)
context.getStateStore("my-win-store");
}
@Override
public void process(String key, String value) {
}
@Override
public void close() {
}
}
{code}
{code:java}
package com.cantgetthistowork;
import java.time.Duration;
import java.time.Instant;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.processor.MockProcessorContext;
import org.apache.kafka.streams.state.Stores;
import org.apache.kafka.streams.state.WindowStore;
import org.junit.Before;
import org.junit.Test;
public class InMemWindowProcessorTest {
InMemWindowProcessor processor = null;
MockProcessorContext context = null;
@Before
public void setup() {
processor = new InMemWindowProcessor();
context = new MockProcessorContext();
WindowStore store =
Stores.windowStoreBuilder(
Stores.inMemoryWindowStore(
"my-win-store",
Duration.ofMinutes(10),
Duration.ofSeconds(10),
false
),
Serdes.String(),
Serdes.String()
)
.withLoggingDisabled()
.build();
store.init(context, store);
context.register(store, null);
processor.init(context);
}
@Test
public void testThings() {
Instant baseTime = Instant.now();
context.setTimestamp(baseTime.toEpochMilli());
context.setTopic("topic-name");
processor.process("key1", "value1");
}
}
{code}
I was trying this with maven, with mvn --version outputting:
{noformat}
Apache Maven 3.5.0 (ff8f5e7444045639af65f6095c62210b5713f426;
2017-04-03T13:39:06-06:00)
Maven home: ~/opt/apache-maven-3.5.0
Java version: 1.8.0_212, vendor: Oracle Corporation
Java home: /usr/lib/jvm/java-8-openjdk-amd64/jre
Default locale: en_US, platform encoding: UTF-8
OS name: "linux", version: "4.15.0-52-generic", arch: "amd64", family:
"unix"{noformat}
And finally the stack trace:
{noformat}
---
T E S T S
---
Running com.cantgetthistowork.InMemWindowProcessorTest
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further
details.
Tests run: 1, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 0.076 sec <<<
FAILURE!
testThings(com.cantgetthistowork.InMemWindowProcessorTest) Time elapsed: 0.05
sec <<< ERROR!
java.lang.ClassCastException:
org.apache.kafka.streams.processor.MockProcessorContext cannot be cast to
org.apache.kafka.streams.processor.internals.InternalProcessorContext
at
org.apache.kafka.streams.state.internals.InMemoryWindowStore.init(InMemoryWindowStore.java:91)
at
org.apache.kafka.streams.state.internals.WrappedStateStore.init(WrappedStateStore.java:48)
at
org.apache.kafka.streams.state.internals.MeteredWindowStore.init(MeteredWindowStore.java:90)
at
com.cantgetthistowork.InMemWindowProcessorTest.setup(InMemWindowProcessorTest.java:36)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at
org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
at
org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
at
org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
at
org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:24)
at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
at
org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
at
org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
at org.junit.runners.ParentRunner$3.run(P
[jira] [Created] (KAFKA-8630) Unit testing a streams processor with a WindowStore throws a ClassCastException
Justin Fetherolf created KAFKA-8630:
---
Summary: Unit testing a streams processor with a WindowStore
throws a ClassCastException
Key: KAFKA-8630
URL: https://issues.apache.org/jira/browse/KAFKA-8630
Project: Kafka
Issue Type: Bug
Components: streams
Affects Versions: 2.3.0
Reporter: Justin Fetherolf
I was attempting to write a unit test for a class implementing the
{{Processor}} interface that contained a {{WindowStore}}, but running the test
fails with a {{ClassCastException}} coming out of {{InMemoryWindowStore.init}}
attempting to cast {{MockProcessorContext}} to {{InternalProcessorContext}}.
Minimal code to reproduce:
{{package com.cantgetthistowork;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.state.WindowStore;
public class InMemWindowProcessor implements Processor \{
private ProcessorContext context;
private WindowStore windowStore;
@Override
public void init(ProcessorContext context) {
this.context = context;
windowStore = (WindowStore)
context.getStateStore("my-win-store");
}
@Override
public void process(String key, String value) \{
}
@Override
public void close() \{
}
}}}
{code:java}
package com.cantgetthistowork;
import java.time.Duration;
import java.time.Instant;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.processor.MockProcessorContext;
import org.apache.kafka.streams.state.Stores;
import org.apache.kafka.streams.state.WindowStore;
import org.junit.Before;
import org.junit.Test;
public class InMemWindowProcessorTest {
InMemWindowProcessor processor = null;
MockProcessorContext context = null;
@Before
public void setup() {
processor = new InMemWindowProcessor();
context = new MockProcessorContext();
WindowStore store =
Stores.windowStoreBuilder(
Stores.inMemoryWindowStore(
"my-win-store",
Duration.ofMinutes(10),
Duration.ofSeconds(10),
false
),
Serdes.String(),
Serdes.String()
)
.withLoggingDisabled()
.build();
store.init(context, store);
context.register(store, null);
processor.init(context);
}
@Test
public void testThings() {
Instant baseTime = Instant.now();
context.setTimestamp(baseTime.toEpochMilli());
context.setTopic("topic-name");
processor.process("key1", "value1");
}
}
{code}
I was trying this with maven, with mvn --version outputting:
{noformat}
Apache Maven 3.5.0 (ff8f5e7444045639af65f6095c62210b5713f426;
2017-04-03T13:39:06-06:00)
Maven home: ~/opt/apache-maven-3.5.0
Java version: 1.8.0_212, vendor: Oracle Corporation
Java home: /usr/lib/jvm/java-8-openjdk-amd64/jre
Default locale: en_US, platform encoding: UTF-8
OS name: "linux", version: "4.15.0-52-generic", arch: "amd64", family:
"unix"{noformat}
And finally the stack trace:
{noformat}
---
T E S T S
---
Running com.cantgetthistowork.InMemWindowProcessorTest
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further
details.
Tests run: 1, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 0.076 sec <<<
FAILURE!
testThings(com.cantgetthistowork.InMemWindowProcessorTest) Time elapsed: 0.05
sec <<< ERROR!
java.lang.ClassCastException:
org.apache.kafka.streams.processor.MockProcessorContext cannot be cast to
org.apache.kafka.streams.processor.internals.InternalProcessorContext
at
org.apache.kafka.streams.state.internals.InMemoryWindowStore.init(InMemoryWindowStore.java:91)
at
org.apache.kafka.streams.state.internals.WrappedStateStore.init(WrappedStateStore.java:48)
at
org.apache.kafka.streams.state.internals.MeteredWindowStore.init(MeteredWindowStore.java:90)
at
com.cantgetthistowork.InMemWindowProcessorTest.setup(InMemWindowProcessorTest.java:36)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at
org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
at
org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
at
org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
at
org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:24)
at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
at
or
[jira] [Updated] (KAFKA-8555) Flaky test ExampleConnectIntegrationTest#testSourceConnector
[ https://issues.apache.org/jira/browse/KAFKA-8555?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Bruno Cadonna updated KAFKA-8555: - Attachment: log-job23215.txt > Flaky test ExampleConnectIntegrationTest#testSourceConnector > > > Key: KAFKA-8555 > URL: https://issues.apache.org/jira/browse/KAFKA-8555 > Project: Kafka > Issue Type: Bug >Reporter: Boyang Chen >Priority: Major > Attachments: log-job23145.txt, log-job23215.txt > > > [https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/22798/console] > *02:03:21* > org.apache.kafka.connect.integration.ExampleConnectIntegrationTest.testSourceConnector > failed, log available in > /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk8-scala2.11/connect/runtime/build/reports/testOutput/org.apache.kafka.connect.integration.ExampleConnectIntegrationTest.testSourceConnector.test.stdout*02:03:21* > *02:03:21* > org.apache.kafka.connect.integration.ExampleConnectIntegrationTest > > testSourceConnector FAILED*02:03:21* > org.apache.kafka.connect.errors.DataException: Insufficient records committed > by connector simple-conn in 15000 millis. Records expected=2000, > actual=1013*02:03:21* at > org.apache.kafka.connect.integration.ConnectorHandle.awaitCommits(ConnectorHandle.java:188)*02:03:21* > at > org.apache.kafka.connect.integration.ExampleConnectIntegrationTest.testSourceConnector(ExampleConnectIntegrationTest.java:181)*02:03:21* -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-8555) Flaky test ExampleConnectIntegrationTest#testSourceConnector
[
https://issues.apache.org/jira/browse/KAFKA-8555?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16878601#comment-16878601
]
Bruno Cadonna commented on KAFKA-8555:
--
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/23215
{code:java}
org.apache.kafka.connect.errors.DataException: Insufficient records committed
by connector simple-conn in 15000 millis. Records expected=2000, actual=1185
at
org.apache.kafka.connect.integration.ConnectorHandle.awaitCommits(ConnectorHandle.java:188)
at
org.apache.kafka.connect.integration.ExampleConnectIntegrationTest.testSourceConnector(ExampleConnectIntegrationTest.java:181)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at
org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59)
{code}
log attached: log-job23215.txt
> Flaky test ExampleConnectIntegrationTest#testSourceConnector
>
>
> Key: KAFKA-8555
> URL: https://issues.apache.org/jira/browse/KAFKA-8555
> Project: Kafka
> Issue Type: Bug
>Reporter: Boyang Chen
>Priority: Major
> Attachments: log-job23145.txt
>
>
> [https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/22798/console]
> *02:03:21*
> org.apache.kafka.connect.integration.ExampleConnectIntegrationTest.testSourceConnector
> failed, log available in
> /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk8-scala2.11/connect/runtime/build/reports/testOutput/org.apache.kafka.connect.integration.ExampleConnectIntegrationTest.testSourceConnector.test.stdout*02:03:21*
> *02:03:21*
> org.apache.kafka.connect.integration.ExampleConnectIntegrationTest >
> testSourceConnector FAILED*02:03:21*
> org.apache.kafka.connect.errors.DataException: Insufficient records committed
> by connector simple-conn in 15000 millis. Records expected=2000,
> actual=1013*02:03:21* at
> org.apache.kafka.connect.integration.ConnectorHandle.awaitCommits(ConnectorHandle.java:188)*02:03:21*
> at
> org.apache.kafka.connect.integration.ExampleConnectIntegrationTest.testSourceConnector(ExampleConnectIntegrationTest.java:181)*02:03:21*
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Comment Edited] (KAFKA-5998) /.checkpoint.tmp Not found exception
[
https://issues.apache.org/jira/browse/KAFKA-5998?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16878549#comment-16878549
]
Patrik Kleindl edited comment on KAFKA-5998 at 7/4/19 10:57 AM:
As this is hopefully coming to an end thanks to [~vvcephei] I'll try a little
writeup so others can compare their setup and hopefully verify that the fix
covers all situations.
I'll also attach the code I used to reproduce the issue, should be runnable for
anyone with docker. [^Kafka5998.zip]
Setup:
We have a streams application with two subtopologies where the first one does
not have state and the second one does.
State directories are created for all tasks, but as the first subtopology does
not have state it does not get locked internally and is therefor subject to the
cleanup process.
The "trigger" for the cleanup is the absence of any traffic on a topic
partition which will allow the directory to be deleted, so this has higher
probability on any low-volume topics (in our case, CDC from a reference table
with little to no changes).
After the directory is deleted, the next record received (and any thereafter)
will trigger the warning because Kafka Streams erroneously tries to write a
checkpoint because all tasks tried to do this for all restored partitions, not
only those assigned to them. This is what Johns PR covers.
We have EOS disabled, so it would be good if people could tell if this has
happened with EOS too.
Or if this warning has occurred for anyone with a different topology situation
or on a high volume topic.
For reference our topology:
{code:java}
Topologies:
Sub-topology: 0
Source: KSTREAM-SOURCE-00 (topics: [inputTopic])
--> KSTREAM-FILTER-01
Processor: KSTREAM-FILTER-01 (stores: [])
--> KSTREAM-KEY-SELECT-02
<-- KSTREAM-SOURCE-00
Processor: KSTREAM-KEY-SELECT-02 (stores: [])
--> KSTREAM-FILTER-05
<-- KSTREAM-FILTER-01
Processor: KSTREAM-FILTER-05 (stores: [])
--> KSTREAM-SINK-04
<-- KSTREAM-KEY-SELECT-02
Sink: KSTREAM-SINK-04 (topic: store-repartition)
<-- KSTREAM-FILTER-05
Sub-topology: 1
Source: KSTREAM-SOURCE-06 (topics: [store-repartition])
--> KSTREAM-REDUCE-03
Processor: KSTREAM-REDUCE-03 (stores: [store])
--> KTABLE-TOSTREAM-07
<-- KSTREAM-SOURCE-06
Processor: KTABLE-TOSTREAM-07 (stores: [])
--> KSTREAM-SINK-08
<-- KSTREAM-REDUCE-03
Sink: KSTREAM-SINK-08 (topic: outputTopic)
<-- KTABLE-TOSTREAM-07{code}
was (Author: pkleindl):
As this is hopefully coming to an end thanks to [~vvcephei] I'll try a little
writeup so others can compare their setup and hopefully verify that the fix
covers all situations.
I'll also attach the code I used to reproduce the issue, should be runnable for
anyone with docker.
Setup:
We have a streams application with two subtopologies where the first one does
not have state and the second one does.
State directories are created for all tasks, but as the first subtopology does
not have state it does not get locked internally and is therefor subject to the
cleanup process.
The "trigger" for the cleanup is the absence of any traffic on a topic
partition which will allow the directory to be deleted, so this has higher
probability on any low-volume topics (in our case, CDC from a reference table
with little to no changes).
After the directory is deleted, the next record received (and any thereafter)
will trigger the warning because Kafka Streams erroneously tries to write a
checkpoint because all tasks tried to do this for all restored partitions, not
only those assigned to them. This is what Johns PR covers.
We have EOS disabled, so it would be good if people could tell if this has
happened with EOS too.
Or if this warning has occurred for anyone with a different topology situation
or on a high volume topic.
For reference our topology:
{code:java}
Topologies:
Sub-topology: 0
Source: KSTREAM-SOURCE-00 (topics: [inputTopic])
--> KSTREAM-FILTER-01
Processor: KSTREAM-FILTER-01 (stores: [])
--> KSTREAM-KEY-SELECT-02
<-- KSTREAM-SOURCE-00
Processor: KSTREAM-KEY-SELECT-02 (stores: [])
--> KSTREAM-FILTER-05
<-- KSTREAM-FILTER-01
Processor: KSTREAM-FILTER-05 (stores: [])
--> KSTREAM-SINK-04
<-- KSTREAM-KEY-SELECT-02
Sink: KSTREAM-SINK-04 (topic: store-repartition)
<-- KSTREAM-FILTER-05
Sub-topology: 1
Source: KSTREAM-SOURCE-06 (topics: [store-repartition])
--> KSTREAM-REDUCE-03
Processor: KSTREAM-REDUCE-03 (stores: [store])
--> KTABLE-TOSTREAM-07
<-- KSTREAM-SOURCE-06
Processor: KTABLE-TOSTREAM-07 (stores: [])
--> KSTREAM-SINK-08
<-- KSTR
[jira] [Updated] (KAFKA-5998) /.checkpoint.tmp Not found exception
[
https://issues.apache.org/jira/browse/KAFKA-5998?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Patrik Kleindl updated KAFKA-5998:
--
Attachment: Kafka5998.zip
> /.checkpoint.tmp Not found exception
>
>
> Key: KAFKA-5998
> URL: https://issues.apache.org/jira/browse/KAFKA-5998
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 0.11.0.0, 0.11.0.1, 2.1.1
>Reporter: Yogesh BG
>Assignee: Bill Bejeck
>Priority: Critical
> Attachments: 5998.v1.txt, 5998.v2.txt, Kafka5998.zip, Topology.txt,
> exc.txt, props.txt, streams.txt
>
>
> I have one kafka broker and one kafka stream running... I am running its
> since two days under load of around 2500 msgs per second.. On third day am
> getting below exception for some of the partitions, I have 16 partitions only
> 0_0 and 0_1 gives this error
> {{09:43:25.955 [ks_0_inst-StreamThread-6] WARN
> o.a.k.s.p.i.ProcessorStateManager - Failed to write checkpoint file to
> /data/kstreams/rtp-kafkastreams/0_1/.checkpoint:
> java.io.FileNotFoundException:
> /data/kstreams/rtp-kafkastreams/0_1/.checkpoint.tmp (No such file or
> directory)
> at java.io.FileOutputStream.open(Native Method) ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:221)
> ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:171)
> ~[na:1.7.0_111]
> at
> org.apache.kafka.streams.state.internals.OffsetCheckpoint.write(OffsetCheckpoint.java:73)
> ~[rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.ProcessorStateManager.checkpoint(ProcessorStateManager.java:324)
> ~[rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask$1.run(StreamTask.java:267)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:201)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:260)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:254)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.AssignedTasks$1.apply(AssignedTasks.java:322)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.AssignedTasks.applyToRunningTasks(AssignedTasks.java:415)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.AssignedTasks.commit(AssignedTasks.java:314)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.commitAll(StreamThread.java:700)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:683)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:523)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:480)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:457)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> 09:43:25.974 [ks_0_inst-StreamThread-15] WARN
> o.a.k.s.p.i.ProcessorStateManager - Failed to write checkpoint file to
> /data/kstreams/rtp-kafkastreams/0_0/.checkpoint:
> java.io.FileNotFoundException:
> /data/kstreams/rtp-kafkastreams/0_0/.checkpoint.tmp (No such file or
> directory)
> at java.io.FileOutputStream.open(Native Method) ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:221)
> ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:171)
> ~[na:1.7.0_111]
> at
> org.apache.kafka.streams.state.internals.OffsetCheckpoint.write(OffsetCheckpoint.java:73)
> ~[rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.ProcessorStateManager.checkpoint(ProcessorStateManager.java:324)
> ~[rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask$1.run(StreamTask.java:267)
> [rtp-kafk
[jira] [Commented] (KAFKA-5998) /.checkpoint.tmp Not found exception
[
https://issues.apache.org/jira/browse/KAFKA-5998?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16878549#comment-16878549
]
Patrik Kleindl commented on KAFKA-5998:
---
As this is hopefully coming to an end thanks to [~vvcephei] I'll try a little
writeup so others can compare their setup and hopefully verify that the fix
covers all situations.
I'll also attach the code I used to reproduce the issue, should be runnable for
anyone with docker.
Setup:
We have a streams application with two subtopologies where the first one does
not have state and the second one does.
State directories are created for all tasks, but as the first subtopology does
not have state it does not get locked internally and is therefor subject to the
cleanup process.
The "trigger" for the cleanup is the absence of any traffic on a topic
partition which will allow the directory to be deleted, so this has higher
probability on any low-volume topics (in our case, CDC from a reference table
with little to no changes).
After the directory is deleted, the next record received (and any thereafter)
will trigger the warning because Kafka Streams erroneously tries to write a
checkpoint because all tasks tried to do this for all restored partitions, not
only those assigned to them. This is what Johns PR covers.
We have EOS disabled, so it would be good if people could tell if this has
happened with EOS too.
Or if this warning has occurred for anyone with a different topology situation
or on a high volume topic.
For reference our topology:
{code:java}
Topologies:
Sub-topology: 0
Source: KSTREAM-SOURCE-00 (topics: [inputTopic])
--> KSTREAM-FILTER-01
Processor: KSTREAM-FILTER-01 (stores: [])
--> KSTREAM-KEY-SELECT-02
<-- KSTREAM-SOURCE-00
Processor: KSTREAM-KEY-SELECT-02 (stores: [])
--> KSTREAM-FILTER-05
<-- KSTREAM-FILTER-01
Processor: KSTREAM-FILTER-05 (stores: [])
--> KSTREAM-SINK-04
<-- KSTREAM-KEY-SELECT-02
Sink: KSTREAM-SINK-04 (topic: store-repartition)
<-- KSTREAM-FILTER-05
Sub-topology: 1
Source: KSTREAM-SOURCE-06 (topics: [store-repartition])
--> KSTREAM-REDUCE-03
Processor: KSTREAM-REDUCE-03 (stores: [store])
--> KTABLE-TOSTREAM-07
<-- KSTREAM-SOURCE-06
Processor: KTABLE-TOSTREAM-07 (stores: [])
--> KSTREAM-SINK-08
<-- KSTREAM-REDUCE-03
Sink: KSTREAM-SINK-08 (topic: outputTopic)
<-- KTABLE-TOSTREAM-07{code}
> /.checkpoint.tmp Not found exception
>
>
> Key: KAFKA-5998
> URL: https://issues.apache.org/jira/browse/KAFKA-5998
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 0.11.0.0, 0.11.0.1, 2.1.1
>Reporter: Yogesh BG
>Assignee: Bill Bejeck
>Priority: Critical
> Attachments: 5998.v1.txt, 5998.v2.txt, Topology.txt, exc.txt,
> props.txt, streams.txt
>
>
> I have one kafka broker and one kafka stream running... I am running its
> since two days under load of around 2500 msgs per second.. On third day am
> getting below exception for some of the partitions, I have 16 partitions only
> 0_0 and 0_1 gives this error
> {{09:43:25.955 [ks_0_inst-StreamThread-6] WARN
> o.a.k.s.p.i.ProcessorStateManager - Failed to write checkpoint file to
> /data/kstreams/rtp-kafkastreams/0_1/.checkpoint:
> java.io.FileNotFoundException:
> /data/kstreams/rtp-kafkastreams/0_1/.checkpoint.tmp (No such file or
> directory)
> at java.io.FileOutputStream.open(Native Method) ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:221)
> ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:171)
> ~[na:1.7.0_111]
> at
> org.apache.kafka.streams.state.internals.OffsetCheckpoint.write(OffsetCheckpoint.java:73)
> ~[rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.ProcessorStateManager.checkpoint(ProcessorStateManager.java:324)
> ~[rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask$1.run(StreamTask.java:267)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:201)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:260)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:254)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependen
[jira] [Commented] (KAFKA-6582) Partitions get underreplicated, with a single ISR, and doesn't recover. Other brokers do not take over and we need to manually restart the broker.
[
https://issues.apache.org/jira/browse/KAFKA-6582?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16878462#comment-16878462
]
Shiran Azran commented on KAFKA-6582:
-
Hello,
We are also having the same issue in version 2.0.1 in the past several months
since the last upgrade.
Is this issue resolved in version 2.1.1?
Thanks.
> Partitions get underreplicated, with a single ISR, and doesn't recover. Other
> brokers do not take over and we need to manually restart the broker.
> --
>
> Key: KAFKA-6582
> URL: https://issues.apache.org/jira/browse/KAFKA-6582
> Project: Kafka
> Issue Type: Bug
> Components: network
>Affects Versions: 1.0.0
> Environment: Ubuntu 16.04
> Linux kafka04 4.4.0-109-generic #132-Ubuntu SMP Tue Jan 9 19:52:39 UTC 2018
> x86_64 x86_64 x86_64 GNU/Linux
> java version "9.0.1"
> Java(TM) SE Runtime Environment (build 9.0.1+11)
> Java HotSpot(TM) 64-Bit Server VM (build 9.0.1+11, mixed mode)
> but also tried with the latest JVM 8 before with the same result.
>Reporter: Jurriaan Pruis
>Priority: Major
> Attachments: Screenshot 2019-01-18 at 13.08.17.png, Screenshot
> 2019-01-18 at 13.16.59.png
>
>
> Partitions get underreplicated, with a single ISR, and doesn't recover. Other
> brokers do not take over and we need to manually restart the 'single ISR'
> broker (if you describe the partitions of replicated topic it is clear that
> some partitions are only in sync on this broker).
> This bug resembles KAFKA-4477 a lot, but since that issue is marked as
> resolved this is probably something else but similar.
> We have the same issue (or at least it looks pretty similar) on Kafka 1.0.
> Since upgrading to Kafka 1.0 in November 2017 we've had these issues (we've
> upgraded from Kafka 0.10.2.1).
> This happens almost every 24-48 hours on a random broker. This is why we
> currently have a cronjob which restarts every broker every 24 hours.
> During this issue the ISR shows the following server log:
> {code:java}
> [2018-02-20 12:02:08,342] WARN Attempting to send response via channel for
> which there is no open connection, connection id
> 10.132.0.32:9092-10.14.148.20:56352-96708 (kafka.network.Processor)
> [2018-02-20 12:02:08,364] WARN Attempting to send response via channel for
> which there is no open connection, connection id
> 10.132.0.32:9092-10.14.150.25:54412-96715 (kafka.network.Processor)
> [2018-02-20 12:02:08,349] WARN Attempting to send response via channel for
> which there is no open connection, connection id
> 10.132.0.32:9092-10.14.149.18:35182-96705 (kafka.network.Processor)
> [2018-02-20 12:02:08,379] WARN Attempting to send response via channel for
> which there is no open connection, connection id
> 10.132.0.32:9092-10.14.150.25:54456-96717 (kafka.network.Processor)
> [2018-02-20 12:02:08,448] WARN Attempting to send response via channel for
> which there is no open connection, connection id
> 10.132.0.32:9092-10.14.159.20:36388-96720 (kafka.network.Processor)
> [2018-02-20 12:02:08,683] WARN Attempting to send response via channel for
> which there is no open connection, connection id
> 10.132.0.32:9092-10.14.157.110:41922-96740 (kafka.network.Processor)
> {code}
> Also on the ISR broker, the controller log shows this:
> {code:java}
> [2018-02-20 12:02:14,927] INFO [Controller-3-to-broker-3-send-thread]:
> Controller 3 connected to 10.132.0.32:9092 (id: 3 rack: null) for sending
> state change requests (kafka.controller.RequestSendThread)
> [2018-02-20 12:02:14,927] INFO [Controller-3-to-broker-0-send-thread]:
> Controller 3 connected to 10.132.0.10:9092 (id: 0 rack: null) for sending
> state change requests (kafka.controller.RequestSendThread)
> [2018-02-20 12:02:14,928] INFO [Controller-3-to-broker-1-send-thread]:
> Controller 3 connected to 10.132.0.12:9092 (id: 1 rack: null) for sending
> state change requests (kafka.controller.RequestSendThread){code}
> And the non-ISR brokers show these kind of errors:
>
> {code:java}
> 2018-02-20 12:02:29,204] WARN [ReplicaFetcher replicaId=1, leaderId=3,
> fetcherId=0] Error in fetch to broker 3, request (type=FetchRequest,
> replicaId=1, maxWait=500, minBytes=1, maxBytes=10485760,
> fetchData={..}, isolationLevel=READ_UNCOMMITTED)
> (kafka.server.ReplicaFetcherThread)
> java.io.IOException: Connection to 3 was disconnected before the response was
> read
> at
> org.apache.kafka.clients.NetworkClientUtils.sendAndReceive(NetworkClientUtils.java:95)
> at
> kafka.server.ReplicaFetcherBlockingSend.sendRequest(ReplicaFetcherBlockingSend.scala:96)
> at kafka.server.ReplicaFetcherThread.fetch(ReplicaFetcherThread.scala:205)
> at kafka.server.Replica
[jira] [Commented] (KAFKA-8106) Reducing the allocation and copying of ByteBuffer when logValidator do validation.
[ https://issues.apache.org/jira/browse/KAFKA-8106?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16878420#comment-16878420 ] ASF GitHub Bot commented on KAFKA-8106: --- hachikuji commented on pull request #6699: KAFKA-8106:Reducing the allocation and copying of ByteBuffer when logValidator do validation(target trunk). URL: https://github.com/apache/kafka/pull/6699 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Reducing the allocation and copying of ByteBuffer when logValidator do > validation. > > > Key: KAFKA-8106 > URL: https://issues.apache.org/jira/browse/KAFKA-8106 > Project: Kafka > Issue Type: Bug > Components: core >Affects Versions: 2.2.0, 2.1.1 > Environment: Server : > cpu:2*16 ; > MemTotal : 256G; > Ethernet controller:Intel Corporation 82599ES 10-Gigabit SFI/SFP+ Network > Connection ; > SSD. >Reporter: Flower.min >Assignee: Flower.min >Priority: Major > Labels: performance > Fix For: 2.4.0 > > > We do performance testing about Kafka in specific scenarios as > described below .We build a kafka cluster with one broker,and create topics > with different number of partitions.Then we start lots of producer processes > to send large amounts of messages to one of the topics at one testing . > *_Specific Scenario_* > > *_1.Main config of Kafka_* > # Main config of Kafka > server:[num.network.threads=6;num.io.threads=128;queued.max.requests=500|http://num.network.threads%3D6%3Bnum.io.threads%3D128%3Bqueued.max.requests%3D500/] > # Number of TopicPartition : 50~2000 > # Size of Single Message : 1024B > > *_2.Config of KafkaProducer_* > ||compression.type||[linger.ms|http://linger.ms/]||batch.size||buffer.memory|| > |lz4|1000ms~5000ms|16KB/10KB/100KB|128MB| > *_3.The best result of performance testing_* > ||Network inflow rate||CPU Used (%)||Disk write speed||Performance of > production|| > |550MB/s~610MB/s|97%～99%|550MB/s~610MB/s |23,000,000 messages/s| > *_4.Phenomenon and my doubt_* > _The upper limit of CPU usage has been reached But it does not > reach the upper limit of the bandwidth of the server network. *We are > doubtful about which cost too much CPU time and we want to Improve > performance and reduces CPU usage of Kafka server.*_ > _*5.Analysis*_ > We analysis the JFIR of Kafka server when doing performance testing > .After we checked and completed the performance test again, we located the > code "*ByteBuffer recordBuffer = > ByteBuffer.allocate(sizeOfBodyInBytes);*(*Class:DefaultRecord,Function:readFrom()*)” > which consumed CPU resources and caused a lot of GC .Our modified code > reduces the allocation and copying of ByteBuffer, so the test performance is > greatly improved, and the CPU's stable usage is *below 60%*. The following is > a comparison of different code test performance under the same conditions. > *Result of performance testing* > *Main config of Kafka: Single > Message:1024B;TopicPartitions:200;linger.ms:1000ms.* > | Single Message : 1024B,|Network inflow rate|CPU(%)|Messages/s| > |Source code|600M/s|97%|25,000,000| > |Modified code|1GB/s|<60%|41,660,000| > **1.Before modified code(Source code) GC:** >  > **2.After modified code(remove allocation of ByteBuffer) GC:** >  -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-8624) 这边是服务端的kafka，版本是 kafka_2.11-1.0.0.2，当客户端向主题发送消息的时候抛出如下异常，客户端跟服务端的版本不一致吗？
[
https://issues.apache.org/jira/browse/KAFKA-8624?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16878406#comment-16878406
]
CHARELS commented on KAFKA-8624:
Thanks for your answer，we'll try
> 这边是服务端的kafka，版本是 kafka_2.11-1.0.0.2，当客户端向主题发送消息的时候抛出如下异常，客户端跟服务端的版本不一致吗？
>
>
> Key: KAFKA-8624
> URL: https://issues.apache.org/jira/browse/KAFKA-8624
> Project: Kafka
> Issue Type: Bug
> Components: log
>Affects Versions: 1.0.0
>Reporter: CHARELS
>Priority: Major
> Labels: pull-request-available
> Attachments: image-2019-07-04-00-18-15-781.png,
> image-2019-07-04-14-49-47-802.png, image-2019-07-04-14-56-24-216.png, 李志涛.jpg
>
> Original Estimate: 12h
> Remaining Estimate: 12h
>
> ERROR [KafkaApi-1004] Error when handling request
> \{replica_id=-1,max_wait_time=1,min_bytes=0,topics=[{topic=ETE,partitions=[{partition=3,fetch_offset=119224671,max_bytes=1048576}]}]}
> (kafka.server.KafkaApis)
> java.lang.IllegalArgumentException: Magic v0 does not support record headers
> at
> org.apache.kafka.common.record.MemoryRecordsBuilder.appendWithOffset(MemoryRecordsBuilder.java:403)
> at
> org.apache.kafka.common.record.MemoryRecordsBuilder.append(MemoryRecordsBuilder.java:586)
> at
> org.apache.kafka.common.record.AbstractRecords.convertRecordBatch(AbstractRecords.java:134)
> at
> org.apache.kafka.common.record.AbstractRecords.downConvert(AbstractRecords.java:109)
> at
> org.apache.kafka.common.record.FileRecords.downConvert(FileRecords.java:253)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$convertedPartitionData$1$1$$anonfun$apply$4.apply(KafkaApis.scala:519)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$convertedPartitionData$1$1$$anonfun$apply$4.apply(KafkaApis.scala:517)
> at scala.Option.map(Option.scala:146)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$convertedPartitionData$1$1.apply(KafkaApis.scala:517)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$convertedPartitionData$1$1.apply(KafkaApis.scala:507)
> at scala.Option.flatMap(Option.scala:171)
> at
> kafka.server.KafkaApis.kafka$server$KafkaApis$$convertedPartitionData$1(KafkaApis.scala:507)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$createResponse$2$1.apply(KafkaApis.scala:555)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$createResponse$2$1.apply(KafkaApis.scala:554)
> 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.server.KafkaApis.kafka$server$KafkaApis$$createResponse$2(KafkaApis.scala:554)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$fetchResponseCallback$1$1.apply(KafkaApis.scala:568)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$fetchResponseCallback$1$1.apply(KafkaApis.scala:568)
> at
> kafka.server.KafkaApis$$anonfun$sendResponseMaybeThrottle$1.apply$mcVI$sp(KafkaApis.scala:2033)
> at
> kafka.server.ClientRequestQuotaManager.maybeRecordAndThrottle(ClientRequestQuotaManager.scala:54)
> at kafka.server.KafkaApis.sendResponseMaybeThrottle(KafkaApis.scala:2032)
> at
> kafka.server.KafkaApis.kafka$server$KafkaApis$$fetchResponseCallback$1(KafkaApis.scala:568)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$processResponseCallback$1$1.apply$mcVI$sp(KafkaApis.scala:587)
> at
> kafka.server.ClientQuotaManager.maybeRecordAndThrottle(ClientQuotaManager.scala:176)
> at
> kafka.server.KafkaApis.kafka$server$KafkaApis$$processResponseCallback$1(KafkaApis.scala:586)
> at
> kafka.server.KafkaApis$$anonfun$handleFetchRequest$3.apply(KafkaApis.scala:603)
> at
> kafka.server.KafkaApis$$anonfun$handleFetchRequest$3.apply(KafkaApis.scala:603)
> at kafka.server.ReplicaManager.fetchMessages(ReplicaManager.scala:820)
> at kafka.server.KafkaApis.handleFetchRequest(KafkaApis.scala:595)
> at kafka.server.KafkaApis.handle(KafkaApis.scala:99)
> at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:65)
> at java.lang.Thread.run(Thread.java:748)
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Comment Edited] (KAFKA-8624) 这边是服务端的kafka，版本是 kafka_2.11-1.0.0.2，当客户端向主题发送消息的时候抛出如下异常，客户端跟服务端的版本不一致吗？
[
https://issues.apache.org/jira/browse/KAFKA-8624?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16878387#comment-16878387
]
Zhitao Li(李志涛) edited comment on KAFKA-8624 at 7/4/19 7:22 AM:
---
!image-2019-07-04-14-56-24-216.png|width=560,height=380!
There are two ways to solve this problem：
One the hig producer version degrade to 0.9.x，another the low consumer version
upgrade to 1.0.x which enabled it support to 'MAGIC_VALUE_V2' of 'RecordBatch'
was (Author: lizhitao):
!image-2019-07-04-14-56-24-216.png|width=600,height=380!
There are two ways to solve this problem：
One the hig producer version degrade to 0.9.x，another the low consumer version
upgrade to 1.0.x which enabled it support to 'MAGIC_VALUE_V2' of 'RecordBatch'
> 这边是服务端的kafka，版本是 kafka_2.11-1.0.0.2，当客户端向主题发送消息的时候抛出如下异常，客户端跟服务端的版本不一致吗？
>
>
> Key: KAFKA-8624
> URL: https://issues.apache.org/jira/browse/KAFKA-8624
> Project: Kafka
> Issue Type: Bug
> Components: log
>Affects Versions: 1.0.0
>Reporter: CHARELS
>Priority: Major
> Labels: pull-request-available
> Attachments: image-2019-07-04-00-18-15-781.png,
> image-2019-07-04-14-49-47-802.png, image-2019-07-04-14-56-24-216.png, 李志涛.jpg
>
> Original Estimate: 12h
> Remaining Estimate: 12h
>
> ERROR [KafkaApi-1004] Error when handling request
> \{replica_id=-1,max_wait_time=1,min_bytes=0,topics=[{topic=ETE,partitions=[{partition=3,fetch_offset=119224671,max_bytes=1048576}]}]}
> (kafka.server.KafkaApis)
> java.lang.IllegalArgumentException: Magic v0 does not support record headers
> at
> org.apache.kafka.common.record.MemoryRecordsBuilder.appendWithOffset(MemoryRecordsBuilder.java:403)
> at
> org.apache.kafka.common.record.MemoryRecordsBuilder.append(MemoryRecordsBuilder.java:586)
> at
> org.apache.kafka.common.record.AbstractRecords.convertRecordBatch(AbstractRecords.java:134)
> at
> org.apache.kafka.common.record.AbstractRecords.downConvert(AbstractRecords.java:109)
> at
> org.apache.kafka.common.record.FileRecords.downConvert(FileRecords.java:253)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$convertedPartitionData$1$1$$anonfun$apply$4.apply(KafkaApis.scala:519)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$convertedPartitionData$1$1$$anonfun$apply$4.apply(KafkaApis.scala:517)
> at scala.Option.map(Option.scala:146)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$convertedPartitionData$1$1.apply(KafkaApis.scala:517)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$convertedPartitionData$1$1.apply(KafkaApis.scala:507)
> at scala.Option.flatMap(Option.scala:171)
> at
> kafka.server.KafkaApis.kafka$server$KafkaApis$$convertedPartitionData$1(KafkaApis.scala:507)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$createResponse$2$1.apply(KafkaApis.scala:555)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$createResponse$2$1.apply(KafkaApis.scala:554)
> 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.server.KafkaApis.kafka$server$KafkaApis$$createResponse$2(KafkaApis.scala:554)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$fetchResponseCallback$1$1.apply(KafkaApis.scala:568)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$fetchResponseCallback$1$1.apply(KafkaApis.scala:568)
> at
> kafka.server.KafkaApis$$anonfun$sendResponseMaybeThrottle$1.apply$mcVI$sp(KafkaApis.scala:2033)
> at
> kafka.server.ClientRequestQuotaManager.maybeRecordAndThrottle(ClientRequestQuotaManager.scala:54)
> at kafka.server.KafkaApis.sendResponseMaybeThrottle(KafkaApis.scala:2032)
> at
> kafka.server.KafkaApis.kafka$server$KafkaApis$$fetchResponseCallback$1(KafkaApis.scala:568)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$processResponseCallback$1$1.apply$mcVI$sp(KafkaApis.scala:587)
> at
> kafka.server.ClientQuotaManager.maybeRecordAndThrottle(ClientQuotaManager.scala:176)
> at
> kafka.server.KafkaApis.kafka$server$KafkaApis$$processResponseCallback$1(KafkaApis.scala:586)
> at
> kafka.server.KafkaApis$$anonfun$handleFetchRequest$3.apply(KafkaApis.scala:603)
> at
> kafka.server.KafkaApis$$anonfun$handleFetchRequest$3.apply(KafkaApis.scala:603)
> at kafka.server.ReplicaManager.fetchMessages(ReplicaManager.scala:820)
> at kafka.server.KafkaApis.handleFetchRequest(KafkaApis.scala:595)
> at kafka.server.KafkaApis.handle(KafkaApis.scala:99)
> at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:65
[jira] [Commented] (KAFKA-8624) 这边是服务端的kafka，版本是 kafka_2.11-1.0.0.2，当客户端向主题发送消息的时候抛出如下异常，客户端跟服务端的版本不一致吗？
[
https://issues.apache.org/jira/browse/KAFKA-8624?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16878387#comment-16878387
]
Zhitao Li(李志涛) commented on KAFKA-8624:
---
!image-2019-07-04-14-56-24-216.png|width=600,height=380!
There are two ways to solve this problem：
One the hig producer version degrade to 0.9.x，another the low consumer version
upgrade to 1.0.x which enabled it support to 'MAGIC_VALUE_V2' of 'RecordBatch'
> 这边是服务端的kafka，版本是 kafka_2.11-1.0.0.2，当客户端向主题发送消息的时候抛出如下异常，客户端跟服务端的版本不一致吗？
>
>
> Key: KAFKA-8624
> URL: https://issues.apache.org/jira/browse/KAFKA-8624
> Project: Kafka
> Issue Type: Bug
> Components: log
>Affects Versions: 1.0.0
>Reporter: CHARELS
>Priority: Major
> Labels: pull-request-available
> Attachments: image-2019-07-04-00-18-15-781.png,
> image-2019-07-04-14-49-47-802.png, image-2019-07-04-14-56-24-216.png, 李志涛.jpg
>
> Original Estimate: 12h
> Remaining Estimate: 12h
>
> ERROR [KafkaApi-1004] Error when handling request
> \{replica_id=-1,max_wait_time=1,min_bytes=0,topics=[{topic=ETE,partitions=[{partition=3,fetch_offset=119224671,max_bytes=1048576}]}]}
> (kafka.server.KafkaApis)
> java.lang.IllegalArgumentException: Magic v0 does not support record headers
> at
> org.apache.kafka.common.record.MemoryRecordsBuilder.appendWithOffset(MemoryRecordsBuilder.java:403)
> at
> org.apache.kafka.common.record.MemoryRecordsBuilder.append(MemoryRecordsBuilder.java:586)
> at
> org.apache.kafka.common.record.AbstractRecords.convertRecordBatch(AbstractRecords.java:134)
> at
> org.apache.kafka.common.record.AbstractRecords.downConvert(AbstractRecords.java:109)
> at
> org.apache.kafka.common.record.FileRecords.downConvert(FileRecords.java:253)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$convertedPartitionData$1$1$$anonfun$apply$4.apply(KafkaApis.scala:519)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$convertedPartitionData$1$1$$anonfun$apply$4.apply(KafkaApis.scala:517)
> at scala.Option.map(Option.scala:146)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$convertedPartitionData$1$1.apply(KafkaApis.scala:517)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$convertedPartitionData$1$1.apply(KafkaApis.scala:507)
> at scala.Option.flatMap(Option.scala:171)
> at
> kafka.server.KafkaApis.kafka$server$KafkaApis$$convertedPartitionData$1(KafkaApis.scala:507)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$createResponse$2$1.apply(KafkaApis.scala:555)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$createResponse$2$1.apply(KafkaApis.scala:554)
> 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.server.KafkaApis.kafka$server$KafkaApis$$createResponse$2(KafkaApis.scala:554)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$fetchResponseCallback$1$1.apply(KafkaApis.scala:568)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$fetchResponseCallback$1$1.apply(KafkaApis.scala:568)
> at
> kafka.server.KafkaApis$$anonfun$sendResponseMaybeThrottle$1.apply$mcVI$sp(KafkaApis.scala:2033)
> at
> kafka.server.ClientRequestQuotaManager.maybeRecordAndThrottle(ClientRequestQuotaManager.scala:54)
> at kafka.server.KafkaApis.sendResponseMaybeThrottle(KafkaApis.scala:2032)
> at
> kafka.server.KafkaApis.kafka$server$KafkaApis$$fetchResponseCallback$1(KafkaApis.scala:568)
> at
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$processResponseCallback$1$1.apply$mcVI$sp(KafkaApis.scala:587)
> at
> kafka.server.ClientQuotaManager.maybeRecordAndThrottle(ClientQuotaManager.scala:176)
> at
> kafka.server.KafkaApis.kafka$server$KafkaApis$$processResponseCallback$1(KafkaApis.scala:586)
> at
> kafka.server.KafkaApis$$anonfun$handleFetchRequest$3.apply(KafkaApis.scala:603)
> at
> kafka.server.KafkaApis$$anonfun$handleFetchRequest$3.apply(KafkaApis.scala:603)
> at kafka.server.ReplicaManager.fetchMessages(ReplicaManager.scala:820)
> at kafka.server.KafkaApis.handleFetchRequest(KafkaApis.scala:595)
> at kafka.server.KafkaApis.handle(KafkaApis.scala:99)
> at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:65)
> at java.lang.Thread.run(Thread.java:748)
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (KAFKA-8621) KIP-486: Support custom way to load KeyStore and TrustStore
[ https://issues.apache.org/jira/browse/KAFKA-8621?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maulin Vasavada updated KAFKA-8621: --- Summary: KIP-486: Support custom way to load KeyStore and TrustStore (was: KIP-486: Support for pluggable KeyStore and TrustStore) > KIP-486: Support custom way to load KeyStore and TrustStore > --- > > Key: KAFKA-8621 > URL: https://issues.apache.org/jira/browse/KAFKA-8621 > Project: Kafka > Issue Type: New Feature > Components: security >Reporter: Maulin Vasavada >Assignee: Thomas Zhou >Priority: Minor > > [https://cwiki.apache.org/confluence/display/KAFKA/KIP-486%3A+Support+for+pluggable+KeyStore+and+TrustStore] -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-8621) KIP-486: Support custom way to load KeyStore and TrustStore
[ https://issues.apache.org/jira/browse/KAFKA-8621?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maulin Vasavada updated KAFKA-8621: --- Description: https://cwiki.apache.org/confluence/display/KAFKA/KIP-486%3A+Support+custom+way+to+load+KeyStore+and+TrustStore (was: [https://cwiki.apache.org/confluence/display/KAFKA/KIP-486%3A+Support+for+pluggable+KeyStore+and+TrustStore]) > KIP-486: Support custom way to load KeyStore and TrustStore > --- > > Key: KAFKA-8621 > URL: https://issues.apache.org/jira/browse/KAFKA-8621 > Project: Kafka > Issue Type: New Feature > Components: security >Reporter: Maulin Vasavada >Assignee: Thomas Zhou >Priority: Minor > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-486%3A+Support+custom+way+to+load+KeyStore+and+TrustStore -- This message was sent by Atlassian JIRA (v7.6.3#76005)
