Re: Regarding Producer Metric - "request-latency-avg"

[Philip Nee]

Hey Amrit,

Could you elaborate on "this is not updated in producer metrics
documentation in confluent"? The confluent documentation
states,
"The average request latency in ms", which seems right to me.

If you trace the code, the latency measures the time difference between the
time the request was created and the response received. I believe it is
between the time send() was called, and the time the producer receives the
response from the broker. And as you might know, the producer doesn't
immediately transmit the sends when send() is called, the requests can be
batched depending on your configuration.

Hope this helps,
P


On Sat, Nov 5, 2022 at 9:02 AM Amrit Gupta  wrote:

> Bumping this
>
> On Fri, 4 Nov 2022 at 3:26 PM, Amrit Gupta  wrote:
>
> > Hi,
> >
> > There is a producer metric
> > <
> https://github.com/apache/kafka/blob/14c36c7539cf5a7a54c2c7e30e115b3f4c4a9402/clients/src/main/java/org/apache/kafka/clients/producer/internals/SenderMetricsRegistry.java#L92
> >
> >  ( request-latency-avg ), this is not updated in producer metrics
> > documentation in confluent.
> >
> > Also can I please get the detail of this metric that records the time
> > spent from which point to which point exactly?
> >
> > Does it include the time it resides in the producer buffer? or it is from
> > the point it is removed from the producer buffer.
> >
> > I mean does this include the time from application calling send() and the
> > record is sent to the broker?
> >
> > Thanks,
> > Tushar
> >
> >
>

Re: [DISCUSS] KIP-882: Make Kafka Connect REST API request timeouts configurable

[Sagar]

Hey Yash,

Thanks for the explanation. I think it should be fine to delegate the
validation directly to the leader.

Thanks!
Sagar.

On Sat, Nov 5, 2022 at 10:42 AM Yash Mayya  wrote:

> Hi Sagar,
>
> Thanks for chiming in!
>
> > Having said that, why does the worker forward to the
> > leader? I am thinking if the worker can perform the validation on it's
> own,
> > we could let it do the validation instead of forwarding everything to the
> > leader
>
> Only the leader is allowed to perform writes to the config topic, so any
> request that requires a config topic write ends up being forwarded to the
> leader. The `POST /connectors` and `PUT /connectors/{connector}/config`
> endpoints call `Herder::putConnectorConfig` which internally does a config
> validation first before initiating another herder request to write to the
> config topic (in distributed mode). If we want to allow the first worker to
> do the config validation, and then forward the request to the leader just
> for the write to the config topic, we'd either need something like a skip
> validations query parameter on the endpoint like Chris talks about above or
> else a new internal only endpoint that just does a write to the config
> topic without any prior config validation. However, we agreed that this
> optimization doesn't really seem necessary for now and can be done later if
> deemed useful. I'd be happy to hear differing thoughts if any, however.
>
> > I think a bound is certainly needed but IMO it shouldn't go beyond
> > 10 mins considering this is just validation
>
> Yeah, I agree that this seems like a fair value - I've elected to go with a
> default value of 10 minutes for the proposed worker configuration that sets
> an upper bound for the timeout query parameter.
>
> Thanks,
> Yash
>
> On Sat, Nov 5, 2022 at 10:30 AM Yash Mayya  wrote:
>
> > Hi Chris,
> >
> > Thanks again for your feedback. I think a worker configuration for the
> > upper bound makes sense - I initially thought we could hardcode it (just
> > like the current request timeout is), but there's no reason to set
> another
> > artificial bound that isn't user configurable which is exactly what we're
> > trying to change here in the first place. I've updated the KIP based on
> all
> > our discussion above.
> >
> > Thanks,
> > Yash
> >
> > On Thu, Nov 3, 2022 at 11:01 PM Sagar  wrote:
> >
> >> Hey Yash,
> >>
> >> Thanks for the KIP! This looks like a useful feature.
> >>
> >> I think the discussion thread already has some great points by Chris.
> Just
> >> a couple of points/clarifications=>
> >>
> >> Regarding, pt#2 , I guess it might be better to forward to the leader as
> >> suggested by Yash. Having said that, why does the worker forward to the
> >> leader? I am thinking if the worker can perform the validation on it's
> >> own,
> >> we could let it do the validation instead of forwarding everything to
> the
> >> leader(even though it might be cheap to forward all requests to the
> >> leader).
> >>
> >> Pt#3 => I think a bound is certainly needed but IMO it shouldn't go
> beyond
> >> 10 mins considering this is just validation. We shouldn't end up in a
> >> situation where a few faulty connectors end up blocking a lot of request
> >> processing threads, so while increasing the config is certainly helpful,
> >> we
> >> shouldn't set too high a value IMO. Of course I am also open to
> >> suggestions
> >> here.
> >>
> >> Thanks!
> >> Sagar.
> >>
> >> On Thu, Nov 3, 2022 at 9:01 PM Chris Egerton 
> >> wrote:
> >>
> >> > Hi Yash,
> >> >
> >> > RE 2: That's a great point about validations already being performed
> by
> >> the
> >> > leader. For completeness's sake, I'd like to note that this only holds
> >> for
> >> > valid configurations; invalid ones are caught right now before being
> >> > forwarded to the leader. Still, I think it's fine to forward to the
> >> leader
> >> > for now and optimize further in the future if necessary. If frequent
> >> > validations are taking place they should be conducted via the `PUT
> >> > /connector-plugins/{pluginName}/config/validate` endpoint, which won't
> >> do
> >> > any forwarding at all.
> >> >
> >> > RE 3: Yes, those endpoints LGTM. And yes, bounds on the timeout also
> >> seem
> >> > reasonable... maybe a low-importance worker property could work for
> >> that?
> >> > Not sure what would make sense for a default; probably somewhere in
> the
> >> > 10-60 minute range but would be interested in others' thoughts.
> >> >
> >> > Thanks for the clarification on the zombie fencing logic. I think we
> >> might
> >> > want to have some more subtle logic around the interaction between
> >> calls to
> >> > Admin::fenceProducers and a worker-level timeout property if we go
> that
> >> > route, but we can cross that particular bridge if we get back to it.
> >> >
> >> > Cheers,
> >> >
> >> > Chris
> >> >
> >> > On Wed, Nov 2, 2022 at 1:48 PM Yash Mayya 
> wrote:
> >> >
> >> > > Hi Chris,
> >> > >
> >> > > Thanks a lot for the super quick re

Re: Regarding Producer Metric - "request-latency-avg"

[Amrit Gupta]

Bumping this

On Fri, 4 Nov 2022 at 3:26 PM, Amrit Gupta  wrote:

> Hi,
>
> There is a producer metric
> 
>  ( request-latency-avg ), this is not updated in producer metrics
> documentation in confluent.
>
> Also can I please get the detail of this metric that records the time
> spent from which point to which point exactly?
>
> Does it include the time it resides in the producer buffer? or it is from
> the point it is removed from the producer buffer.
>
> I mean does this include the time from application calling send() and the
> record is sent to the broker?
>
> Thanks,
> Tushar
>
>

Re: [DISCUSS] KIP-872: Add Serializer#serializeToByteBuffer() to reduce memory copying

[ShunKang Lin]

Hi John,

Thanks for your comments!

For your first question, I see some unit test cases that give us a
ByteBuffer not set to read before calling
`ByteBufferSerializer#serialize(String, ByteBuffer)`, e.g.
`ArticleSerializer`, `AugmentedArticleSerializer`,
`AugmentedCommentSerializer` and `CommentSerializer`. If we don't flip the
ByteBuffer inside the `ByteBufferSerializer#serialize(String, ByteBuffer)`
it will break user code using `ByteBufferSerializer#serialize(String,
ByteBuffer)`, and if we don't flip the ByteBuffer inside
the `ByteBufferSerializer#serializeToByteBuffer(String, ByteBuffer)`, it
will be even more strange to the user, because
`ByteBufferSerializer#serialize(String, ByteBuffer)` and
`ByteBufferSerializer#serializeToByteBuffer(String, ByteBuffer)` require
users use the ByteBufferSerializer in two different ways. So if we think of
`ByteBufferSerialize#serializeToByteBuffer(String, ByteBuffer)` as setting
up a ByteBuffer to read later, is it more acceptable?

For your second question, I plan to ultimately replace byte[] with
ByteBuffer, I will document the intent in your KIP and JavaDocs later.

I will clarify that if a Serializer implements the new method, then the old
one will never be called.

Best,
ShunKang

John Roesler  于2022年11月4日周五 22:42写道：

> Hi ShunKang,
>
> Thanks for the KIP!
>
> I’ve been wanting to transition toward byte buffers for a while, so this
> is a nice start.
>
> I thought it was a bit weird to flip the buffer inside the serializer, but
> I see the existing one already does that. I would have thought it would
> make more sense for the caller to give us a buffer already set up for
> reading. Do you think it makes sense to adopt this pattern for the new
> method?
>
> Do you plan to keep the new methods as optional indefinitely, or do you
> plan to ultimately replace byte[] with ByteBuffer? If it’s the latter, then
> it would be good to document the intent in your KIP and JavaDocs.
>
> It would be good to clarify that if a Serializer implements the new
> method, then the old one will never be called. That way, implementations
> can just throw an exception on that method instead of implementing both.
>
> Thanks again!
> -John
>
> On Wed, Nov 2, 2022, at 20:14, ShunKang Lin wrote:
> > Bump this thread again : )
> >
> > ShunKang Lin 于2022年9月25日 周日23:59写道：
> >
> >> Hi all, I'd like to start a new discussion thread on KIP-872 (Kafka
> >> Client) which proposes that add Serializer#serializeToByteBuffer() to
> >> reduce memory copying.
> >>
> >> KIP:
> >>
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=228495828
> >> Thanks, ShunKang
> >>
>

Jenkins build is still unstable: Kafka » Kafka Branch Builder » 3.3 #113

[Apache Jenkins Server]

See 

Jenkins build is still unstable: Kafka » Kafka Branch Builder » trunk #1338

[Apache Jenkins Server]

See 

[jira] [Created] (KAFKA-14360) Documentation: Streams Security page has broken links

[Adam (Jira)]

Adam created KAFKA-14360:


 Summary: Documentation: Streams Security page has broken links
 Key: KAFKA-14360
 URL: https://issues.apache.org/jira/browse/KAFKA-14360
 Project: Kafka
  Issue Type: Bug
Reporter: Adam


A number of links on the 'Streams Security' page are 404-ing

https://kafka.apache.org/documentation/streams/developer-guide/security.html

* Kafka’s security features 
https://kafka.apache.org/documentation/documentation.html#security
* Java Producer and Consumer API 
https://kafka.apache.org/documentation/clients/index.html#kafka-clients



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

Build failed in Jenkins: Kafka » Kafka Branch Builder » 3.2 #88

[Apache Jenkins Server]

See 


Changes:


--
[...truncated 303177 lines...]
[2022-11-05T14:21:59.042Z] > Task :connect:api:testClasses UP-TO-DATE
[2022-11-05T14:21:59.042Z] > Task :connect:api:testJar
[2022-11-05T14:21:59.042Z] > Task :connect:api:testSrcJar
[2022-11-05T14:21:59.042Z] > Task 
:connect:api:publishMavenJavaPublicationToMavenLocal
[2022-11-05T14:21:59.042Z] > Task :connect:api:publishToMavenLocal
[2022-11-05T14:22:01.104Z] 
[2022-11-05T14:22:01.104Z] > Task :streams:javadoc
[2022-11-05T14:22:01.105Z] 
/home/jenkins/jenkins-agent/workspace/Kafka_kafka_3.2/streams/src/main/java/org/apache/kafka/streams/TopologyConfig.java:58:
 warning - Tag @link: missing '#': "org.apache.kafka.streams.StreamsBuilder()"
[2022-11-05T14:22:01.105Z] 
/home/jenkins/jenkins-agent/workspace/Kafka_kafka_3.2/streams/src/main/java/org/apache/kafka/streams/TopologyConfig.java:58:
 warning - Tag @link: can't find org.apache.kafka.streams.StreamsBuilder() in 
org.apache.kafka.streams.TopologyConfig
[2022-11-05T14:22:01.105Z] 
/home/jenkins/jenkins-agent/workspace/Kafka_kafka_3.2/streams/src/main/java/org/apache/kafka/streams/query/Position.java:44:
 warning - Tag @link: can't find query(Query,
[2022-11-05T14:22:01.105Z]  PositionBound, boolean) in 
org.apache.kafka.streams.processor.StateStore
[2022-11-05T14:22:01.105Z] 
/home/jenkins/jenkins-agent/workspace/Kafka_kafka_3.2/streams/src/main/java/org/apache/kafka/streams/query/QueryResult.java:44:
 warning - Tag @link: can't find query(Query, PositionBound, boolean) in 
org.apache.kafka.streams.processor.StateStore
[2022-11-05T14:22:01.105Z] 
/home/jenkins/jenkins-agent/workspace/Kafka_kafka_3.2/streams/src/main/java/org/apache/kafka/streams/query/QueryResult.java:36:
 warning - Tag @link: can't find query(Query, PositionBound, boolean) in 
org.apache.kafka.streams.processor.StateStore
[2022-11-05T14:22:01.105Z] 
/home/jenkins/jenkins-agent/workspace/Kafka_kafka_3.2/streams/src/main/java/org/apache/kafka/streams/query/QueryResult.java:57:
 warning - Tag @link: can't find query(Query, PositionBound, boolean) in 
org.apache.kafka.streams.processor.StateStore
[2022-11-05T14:22:01.105Z] 
/home/jenkins/jenkins-agent/workspace/Kafka_kafka_3.2/streams/src/main/java/org/apache/kafka/streams/query/QueryResult.java:74:
 warning - Tag @link: can't find query(Query, PositionBound, boolean) in 
org.apache.kafka.streams.processor.StateStore
[2022-11-05T14:22:01.105Z] 
/home/jenkins/jenkins-agent/workspace/Kafka_kafka_3.2/streams/src/main/java/org/apache/kafka/streams/query/QueryResult.java:110:
 warning - Tag @link: reference not found: this#getResult()
[2022-11-05T14:22:01.105Z] 
/home/jenkins/jenkins-agent/workspace/Kafka_kafka_3.2/streams/src/main/java/org/apache/kafka/streams/query/QueryResult.java:117:
 warning - Tag @link: reference not found: this#getFailureReason()
[2022-11-05T14:22:01.105Z] 
/home/jenkins/jenkins-agent/workspace/Kafka_kafka_3.2/streams/src/main/java/org/apache/kafka/streams/query/QueryResult.java:117:
 warning - Tag @link: reference not found: this#getFailureMessage()
[2022-11-05T14:22:01.105Z] 
/home/jenkins/jenkins-agent/workspace/Kafka_kafka_3.2/streams/src/main/java/org/apache/kafka/streams/query/QueryResult.java:155:
 warning - Tag @link: reference not found: this#isSuccess()
[2022-11-05T14:22:01.105Z] 
/home/jenkins/jenkins-agent/workspace/Kafka_kafka_3.2/streams/src/main/java/org/apache/kafka/streams/query/QueryResult.java:155:
 warning - Tag @link: reference not found: this#isFailure()
[2022-11-05T14:22:02.135Z] 12 warnings
[2022-11-05T14:22:02.135Z] 
[2022-11-05T14:22:02.135Z] > Task :streams:javadocJar
[2022-11-05T14:22:02.135Z] 
[2022-11-05T14:22:02.135Z] > Task :clients:javadoc
[2022-11-05T14:22:02.136Z] 
/home/jenkins/jenkins-agent/workspace/Kafka_kafka_3.2/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/secured/OAuthBearerLoginCallbackHandler.java:147:
 warning - Tag @link: reference not found: 
[2022-11-05T14:22:03.287Z] 1 warning
[2022-11-05T14:22:04.225Z] 
[2022-11-05T14:22:04.225Z] > Task :clients:javadocJar
[2022-11-05T14:22:04.225Z] 
[2022-11-05T14:22:04.225Z] > Task :clients:srcJar
[2022-11-05T14:22:04.225Z] Execution optimizations have been disabled for task 
':clients:srcJar' to ensure correctness due to the following reasons:
[2022-11-05T14:22:04.225Z]   - Gradle detected a problem with the following 
location: 
'/home/jenkins/jenkins-agent/workspace/Kafka_kafka_3.2/clients/src/generated/java'.
 Reason: Task ':clients:srcJar' uses this output of task 
':clients:processMessages' without declaring an explicit or implicit 
dependency. This can lead to incorrect results being produced, depending on 
what order the tasks are executed. Please refer to 
https://docs.gradle.org/7.3.3/userguide/validation_problems.html#implicit_dependency
 for more details about this problem.
[2022-11-05T14:22:05.255Z] 
[2022-11-05T14:22:05.255Z] > Task

[jira] [Resolved] (KAFKA-14355) Integer overflow occurs in kafka-producer-perf-test.sh (class ProducerProfrmance ).

[Luke Chen (Jira)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-14355?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Luke Chen resolved KAFKA-14355.
---
Fix Version/s: 3.4.0
   Resolution: Fixed

> Integer overflow occurs in kafka-producer-perf-test.sh (class 
> ProducerProfrmance ).
> ---
>
> Key: KAFKA-14355
> URL: https://issues.apache.org/jira/browse/KAFKA-14355
> Project: Kafka
>  Issue Type: Bug
>  Components: tools
>Affects Versions: 3.3.1
>Reporter: Minoru Tomioka
>Assignee: Minoru Tomioka
>Priority: Minor
> Fix For: 3.4.0
>
>
>  
> An integer overflow occurs if a fairly large value is passed to 
> {{{}--num-records{}}}.  It causes a {{{}NegativeArraySizeException error{}}}.
> {code:java}
> ./bin/kafka-producer-perf-test.sh --topic=test --producer-props 
> bootstrap.servers=localhost:9092 --num-records 2000 --throughput 
> 1 --record-size 10
> Exception in thread "main" java.lang.NegativeArraySizeException
>     at 
> org.apache.kafka.tools.ProducerPerformance$Stats.(ProducerPerformance.java:354)
>     at 
> org.apache.kafka.tools.ProducerPerformance.start(ProducerPerformance.java:97)
>     at 
> org.apache.kafka.tools.ProducerPerformance.main(ProducerPerformance.java:52){code}



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

Jenkins build is still unstable: Kafka » Kafka Branch Builder » trunk #1337

[Apache Jenkins Server]

See 

Re: [DISCUSS] KIP-793: Sink Connectors: Support topic-mutating SMTs for async connectors (preCommit users)

[Yash Mayya]

Hi Chris,

Thanks a lot for your inputs!

> would provide a simple, clean interface for developers to determine
> which features are supported by the version of the Connect runtime
> that their plugin has been deployed onto

I do like the idea of having such a public compatibility library - I think
it would remove a lot of restrictions from framework development if it were
to be widely adopted.

> we might consider adding an API to "ack" sink records

I agree that this does seem like a more intuitive and clean API, but I'm
concerned about the backward compatibility headache we'd be imposing on all
existing sink connectors. Connector developers will have to maintain two
separate ways of doing offset management if they want to use the new API
but continue supporting older versions of Kafka Connect.

For now, I've reverted the KIP to the previous iteration which proposed the
addition of a new `SinkRecord` method to obtain the original Kafka topic
pre-transformation. One thing to note is that I've removed the method for
obtaining the original Kafka partition after a cursory search showed that
use cases for partition modifying SMTs are primarily on the source
connector side.

Thanks,
Yash

On Tue, Nov 1, 2022 at 9:22 PM Chris Egerton 
wrote:

> Hi all,
>
> I have more comments I'd like to make on this KIP when I have time (sorry
> for the delay, Yash, and thanks for your patience!), but I did want to
> chime in and say that I'm also not sure about overloading SinkTask::put. I
> share the concerns about creating an intuitive, simple API that Yash has
> raised. In addition, this approach doesn't seem very sustainable--what do
> we do if we encounter another case in the future that would warrant a
> similar solution? We probably don't want to create three, four, etc.
> overloaded variants of the method, each of which would have to be
> implemented by connector developers who want to both leverage the latest
> and greatest connector APIs and maintain compatibility with connect
> Clusters running older versions.
>
> I haven't been able to flesh this out into a design worth publishing in its
> own KIP yet, but one alternative I've pitched to a few people with
> generally positive interest has been to develop an official compatibility
> library for Connect developers. This library would be released as its own
> Maven artifact (separate from connect-api, connect-runtime, etc.) and would
> provide a simple, clean interface for developers to determine which
> features are supported by the version of the Connect runtime that their
> plugin has been deployed onto. Under the hood, this library might use
> reflection to determine whether classes, methods, etc. are available, but
> the developer wouldn't have to do anything more than check (for example)
> `Features.SINK_TASK_ERRANT_RECORD_REPORTER.enabled()` to know at any point
> in the lifetime of their connector/task whether that feature is provided by
> the runtime.
>
> One other high-level comment: this doesn't address every case, but we might
> consider adding an API to "ack" sink records. This could use the
> SubmittedRecords class [1] (with some slight tweaks) under the hood to
> track the latest-acked offset for each topic partition. This way, connector
> developers won't be responsible for tracking offsets at all in their sink
> tasks (eliminating issues with the accuracy of post-transformation T/P/O
> sink record information), and they'll only have to notify the Connect
> framework when a record has been successfully dispatched to the external
> system. This provides a cleaner, friendlier API, and also enables more
> fine-grained metrics like the ones proposed in KIP-767 [2].
>
> [1] -
>
> https://github.com/apache/kafka/blob/9ab140d5419d735baae45aff56ffce7f5622744f/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SubmittedRecords.java
> [2] -
>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-767%3A+Connect+Latency+Metrics
>
> Cheers,
>
> Chris
>
> On Tue, Nov 1, 2022 at 11:21 AM Yash Mayya  wrote:
>
> > Hi Randall,
> >
> > It's been a while for this one but the more I think about it, the more I
> > feel like the current approach with a new overloaded `SinkTask::put`
> method
> > might not be optimal. We're trying to fix a pretty corner case bug here
> > (usage of topic mutating SMTs with sink connectors that do their own
> offset
> > tracking) and I'm not sure that warrants a change to such a central
> > interface method. The new `SinkTask::put` method just seems somewhat odd
> > and it may not be very understandable for a new reader - I don't think
> this
> > should be the case for a public interface method. Furthermore, even with
> > elaborate documentation in place, I'm not sure if it'll be very obvious
> to
> > most people what the purpose of having these two `put` methods is and how
> > they should be used by sink task implementations. What do you think?
> >
> > Thanks,
> > Yash
> >
> > On Mon, Oct 10, 2022 at 9:33 PM Yash Mayya  wrote:

Jenkins build is still unstable: Kafka » Kafka Branch Builder » trunk #1336

[Apache Jenkins Server]

See