Re: [VOTE] KIP-654 Aborting Transaction with non-flushed data should throw a non-fatal Exception

[Gokul Srinivas]

Thank you all. I will update the confluence pages with the status
and raise the PR once it is ready.

-Gokul

On 10-09-2020 03:22, Sophie Blee-Goldman wrote:

+1 from me as well (non-binding)

Gokul, it looks you've now received enough binding votes and the vote has
been
open for sufficiently long. You can conclude the vote and open the PR for
review
when it's ready

Cheers,
Sophie

On Wed, Sep 9, 2020 at 2:47 PM Guozhang Wang  wrote:


+1. Thanks for the KIP Gokul !

Guozhang

On Tue, Sep 8, 2020 at 6:52 PM Matthias J. Sax  wrote:


+1 (binding)

On 9/8/20 2:49 PM, Jason Gustafson wrote:

+1 Thanks for the KIP!

On Thu, Sep 3, 2020 at 12:25 PM Gokul Srinivas 

wrote:

Hi,

I would like to call a vote on the following KIP.

*KIP *-



https://cwiki.apache.org/confluence/display/KAFKA/KIP-654:+Aborted+transaction+with+non-flushed+data+should+throw+a+non-fatal+exception

<


https://cwiki.apache.org/confluence/display/KAFKA/KIP-654:+Aborted+transaction+with+non-flushed+data+should+throw+a+non-fatal+exception

TL;DR: This KIP proposes to throw a new, non-fatal exception whilst
aborting transactions with non-flushed data. This will help users
distinguish non-fatal errors and potentially retry the batch.

Thanks,
-Gokul





--
-- Guozhang

Build failed in Jenkins: Kafka » kafka-trunk-jdk15 #52

[Apache Jenkins Server]

See 


Changes:

[github] KAFKA-8334 Make sure the thread which tries to complete delayed reque… 
(#8657)

[github] MINOR: add ImplicitLinkedHashCollection#moveToEnd (#9269)


--
[...truncated 6.57 MB...]

org.apache.kafka.streams.TopologyTestDriverTest > shouldRespectTaskIdling[Eos 
enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldUseSourceSpecificDeserializers[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldUseSourceSpecificDeserializers[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > shouldReturnAllStores[Eos 
enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > shouldReturnAllStores[Eos 
enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldNotCreateStateDirectoryForStatelessTopology[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldNotCreateStateDirectoryForStatelessTopology[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldApplyGlobalUpdatesCorrectlyInRecursiveTopologies[Eos enabled = false] 
STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldApplyGlobalUpdatesCorrectlyInRecursiveTopologies[Eos enabled = false] 
PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldReturnAllStoresNames[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldReturnAllStoresNames[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPassRecordHeadersIntoSerializersAndDeserializers[Eos enabled = false] 
STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPassRecordHeadersIntoSerializersAndDeserializers[Eos enabled = false] 
PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldProcessConsumerRecordList[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldProcessConsumerRecordList[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldUseSinkSpecificSerializers[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldUseSinkSpecificSerializers[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldFlushStoreForFirstInput[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldFlushStoreForFirstInput[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldProcessFromSourceThatMatchPattern[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldProcessFromSourceThatMatchPattern[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldCaptureSinkTopicNamesIfWrittenInto[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldCaptureSinkTopicNamesIfWrittenInto[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldUpdateStoreForNewKey[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldUpdateStoreForNewKey[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldSendRecordViaCorrectSourceTopicDeprecated[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldSendRecordViaCorrectSourceTopicDeprecated[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPunctuateOnWallClockTime[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPunctuateOnWallClockTime[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > shouldSetRecordMetadata[Eos 
enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > shouldSetRecordMetadata[Eos 
enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldNotUpdateStoreForLargerValue[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldNotUpdateStoreForLargerValue[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldReturnCorrectInMemoryStoreTypeOnly[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldReturnCorrectInMemoryStoreTypeOnly[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > shouldThrowForMissingTime[Eos 
enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > shouldThrowForMissingTime[Eos 
enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldCaptureInternalTopicNamesIfWrittenInto[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldCaptureInternalTopicNamesIfWrittenInto[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 

Re: How do I place a JIRA in status "Patch Available"? (For KAFKA-10473)

[James Cheng]

Thanks John. My wiki user ID is wushujames 

-James

Sent from my iPhone

> On Sep 9, 2020, at 7:03 PM, John Roesler  wrote:
> 
> Hi James,
> 
> Good, I’m glad my incredibly vague response was helpful!
> 
> If you let me know your wiki user id, I can grant you edit permission. It’s a 
> separate account from Jira. 
> 
> Thanks,
> John
> 
>> On Wed, Sep 9, 2020, at 20:45, James Cheng wrote:
>> Thanks John. That worked.
>> 
>> I clicked the button that says "Submit Patch", and a dialog box popped 
>> up. I didn't fill out anything additional in the dialog, and clicked 
>> "Submit Patch" in the dialog.
>> 
>> The JIRA is now in status "Patch Available"
>> 
>> I would like to improve the docs at 
>> https://cwiki.apache.org/confluence/display/KAFKA/Contributing+Code+Changes 
>> 
>>  to make this step clearer. It looks like I don't have permissions to edit 
>> the page.
>> 
>> Can someone grant me permissions to edit the page? 
>> 
>> Or, if that is too difficult, can someone edit the page as follows?
>> 
>> Change
>> 
>>7. Change the status of the JIRA to "Patch Available" if it's ready for 
>> review.
>> to be
>> 
>>7. Change the status of the JIRA to "Patch Available" if it's ready 
>> for review. Do this by clicking the "Submit Patch" button in JIRA, and 
>> then in the resulting dialog, click "Submit Patch".
>> 
>> -James
>> 
 On Sep 9, 2020, at 6:24 PM, John Roesler  wrote:
>>> 
>>> Hi James,
>>> 
>>> I think the button on Jira says “Add Patch” or something confusing like 
>>> that. 
>>> 
>>> Thanks,
>>> John
>>> 
>>> 
>>> On Wed, Sep 9, 2020, at 17:34, James Cheng wrote:
 I have a JIRA that I am working on, and a pull request available for it.
 
 [KAFKA-10473] Website is missing docs on JMX metrics for partition 
 size-on-disk (kafka.log:type=Log,name=*)
 https://issues.apache.org/jira/browse/KAFKA-10473
 https://github.com/apache/kafka/pull/9276
 
 The "Contributing Code Changes" instructions say to
7. Change the status of the JIRA to "Patch Available" if it's ready for 
 review.
 https://cwiki.apache.org/confluence/display/KAFKA/Contributing+Code+Changes
 
 How do I do that? 
 * The title of my pull request starts with KAFKA-10473, so the JIRA 
 does have a link to the pull request
 * I *was* able to assign it to myself and then say "Start progress" and 
 now the status says "In Progress".
 * But I can't find how to set it to "Patch Available". In the JIRA 
 website, I can't find a field or menu item that lets me change the 
 status to "Patch Available" . 
 
 Thanks,
 -James
 
 
 
>> 
>> 

Build failed in Jenkins: Kafka » kafka-trunk-jdk8 #51

[Apache Jenkins Server]

See 


Changes:

[github] MINOR: add ImplicitLinkedHashCollection#moveToEnd (#9269)


--
[...truncated 3.26 MB...]
org.apache.kafka.streams.TopologyTestDriverTest > 
shouldUseSourceSpecificDeserializersDeprecated[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldUseSourceSpecificDeserializersDeprecated[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPunctuateIfEvenTimeAdvances[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPunctuateIfEvenTimeAdvances[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldThrowNoSuchElementExceptionForUnusedOutputTopicWithDynamicRouting[Eos 
enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldThrowNoSuchElementExceptionForUnusedOutputTopicWithDynamicRouting[Eos 
enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > shouldInitProcessor[Eos 
enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > shouldInitProcessor[Eos 
enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldThrowForUnknownTopic[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldThrowForUnknownTopic[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPunctuateOnStreamsTime[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPunctuateOnStreamsTime[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldCaptureGlobalTopicNameIfWrittenInto[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldCaptureGlobalTopicNameIfWrittenInto[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldThrowIfInMemoryBuiltInStoreIsAccessedWithUntypedMethod[Eos enabled = 
false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldThrowIfInMemoryBuiltInStoreIsAccessedWithUntypedMethod[Eos enabled = 
false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldProcessFromSourcesThatMatchMultiplePattern[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldProcessFromSourcesThatMatchMultiplePattern[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > shouldPopulateGlobalStore[Eos 
enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > shouldPopulateGlobalStore[Eos 
enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldThrowIfPersistentBuiltInStoreIsAccessedWithUntypedMethod[Eos enabled = 
false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldThrowIfPersistentBuiltInStoreIsAccessedWithUntypedMethod[Eos enabled = 
false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldAllowPrePopulatingStatesStoresWithCachingEnabled[Eos enabled = false] 
STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldAllowPrePopulatingStatesStoresWithCachingEnabled[Eos enabled = false] 
PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldReturnCorrectPersistentStoreTypeOnly[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldReturnCorrectPersistentStoreTypeOnly[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > shouldRespectTaskIdling[Eos 
enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > shouldRespectTaskIdling[Eos 
enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldUseSourceSpecificDeserializers[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldUseSourceSpecificDeserializers[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > shouldReturnAllStores[Eos 
enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > shouldReturnAllStores[Eos 
enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldNotCreateStateDirectoryForStatelessTopology[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldNotCreateStateDirectoryForStatelessTopology[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldApplyGlobalUpdatesCorrectlyInRecursiveTopologies[Eos enabled = false] 
STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldApplyGlobalUpdatesCorrectlyInRecursiveTopologies[Eos enabled = false] 
PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldReturnAllStoresNames[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldReturnAllStoresNames[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPassRecordHeadersIntoSerializersAndDeserializers[Eos 

Re: [VOTE] KIP-478 Strongly Typed Streams Processor API

[Sophie Blee-Goldman]

>
> If you were to call "put" from a punctuator, or do a
> `range()` query and then update one of those records with
> `put()`, you'd have a very subtle bug on your hands.


Can you elaborate on this a bit? I agree that the punctuator case is an
obvious exemption to the assumption that store invocations always
have a corresponding "current record", but I don't understand the
second example. Are you envisioning a scenario where the #process
method performs a range query and then updates records? Or were
you just giving another example of the punctuator case?

I only bring it up because I agree that the current record information could
still be useful within the context of the store. As a non-user my input on
this
definitely has limited value, but it just isn't striking me as obvious that
we
should remove access to the current record context from the state stores.
If there is no current record, as in the  punctuator case, we should just
set
the record context to null (or Optional.empty, etc).

That said, the put() always has to come from somewhere, and that
somewhere is always going to be either a Processor or a Punctuator, both
of which will still have access to the full context. So additional info
such as
the timestamp can and should probably be supplied to the store before
calling put(), rather than looked up by the store. But I can see some other
things being useful, for example the current record's headers. Maybe if/when
we add better (or any) support for headers in state stores this will be
less true.

Of course as John has made clear, it's pretty hard to judge without
examples
and more insight as to what actually goes on within a custom state store

On Wed, Sep 9, 2020 at 8:07 PM John Roesler  wrote:

> Hi Paul,
>
> It's good to hear from you!
>
> I'm glad you're in favor of the direction. Especially when
> it comes to public API and usability concens, I tend to
> think that "the folks who matter" are actually the folks who
> have to use the APIs to accomplish real tasks. It can be
> hard for me to be sure I'm thinking clearly from that
> perspective.
>
> Funny story, I also started down this road a couple of times
> already and backed them out before the KIP because I was
> afraid of the scope of the proposal. Unfortunately, needing
> to make a new ProcessorContext kind of forced my hand.
>
> I see you've called me out about the ChangeLogging stores :)
> In fact, I think these are the main/only reason that stores
> might really need to invoke "forward()". My secret plan was
> to cheat and either accomplish change-logging by a different
> mechanism than implementing the store interface, or by just
> breaking encapsulation to sneak the "real" ProcessorContext
> into the ChangeLogging stores. But those are all
> implementation details. I think the key question is whether
> anyone else has a store implementation that needs to call
> "forward()". It's not what you mentioned, but since you
> spoke up, I'll just ask: if you have a use case for calling
> "forward()" in a store, please share it.
>
> Regarding the other record-specific context methods, I think
> you have a good point, but I also can't quite wrap my head
> around how we can actually guarantee it to work in general.
> For example, the case you cited, where the implementation of
> `KeyValueStore#put(key, value)` uses the context to augment
> the record with timestamp information. This relies on the
> assumption that you would only call "put()" from inside a
> `Processor#process(key, value)` call in which the record
> being processed is the same record that you're trying to put
> into the store.
>
> If you were to call "put" from a punctuator, or do a
> `range()` query and then update one of those records with
> `put()`, you'd have a very subtle bug on your hands. Right
> now, the Streams component that actually calls the Processor
> takes care to set the right record context before invoking
> the method, and in the case of caching, etc., it also takes
> care to swap out the old context and keep it somewhere safe.
> But when it comes to public API Processors calling methods
> on StateStores, there's no opportunity for any component to
> make sure the context is always correct.
>
> In the face of that situation, it seemed better to just move
> in the direction of a "normal" data store. I.e., when you
> use a HashMap or RocksDB or other "state stores", you don't
> expect them to automatically know extra stuff about the
> record you're storing. If you need them to know something,
> you just put it in the value.
>
> All of that said, I'm just reasoning from first principles
> here. To really know if this is a mistake or not, I need to
> be in your place. So please push back if you think what I
> said is nonsense. My personal plan was to keep an eye out
> during the period where the old API was still present, but
> deprecated, to see if people were struggling to use the new
> API. If so, then we'd have a chance to address it before
> dropping the old 

Re: [VOTE] KIP-478 Strongly Typed Streams Processor API

[John Roesler]

Hi Paul,

It's good to hear from you!

I'm glad you're in favor of the direction. Especially when
it comes to public API and usability concens, I tend to
think that "the folks who matter" are actually the folks who
have to use the APIs to accomplish real tasks. It can be
hard for me to be sure I'm thinking clearly from that
perspective.

Funny story, I also started down this road a couple of times
already and backed them out before the KIP because I was
afraid of the scope of the proposal. Unfortunately, needing
to make a new ProcessorContext kind of forced my hand.

I see you've called me out about the ChangeLogging stores :)
In fact, I think these are the main/only reason that stores
might really need to invoke "forward()". My secret plan was
to cheat and either accomplish change-logging by a different
mechanism than implementing the store interface, or by just
breaking encapsulation to sneak the "real" ProcessorContext
into the ChangeLogging stores. But those are all
implementation details. I think the key question is whether
anyone else has a store implementation that needs to call
"forward()". It's not what you mentioned, but since you
spoke up, I'll just ask: if you have a use case for calling
"forward()" in a store, please share it.

Regarding the other record-specific context methods, I think
you have a good point, but I also can't quite wrap my head
around how we can actually guarantee it to work in general.
For example, the case you cited, where the implementation of
`KeyValueStore#put(key, value)` uses the context to augment
the record with timestamp information. This relies on the
assumption that you would only call "put()" from inside a
`Processor#process(key, value)` call in which the record
being processed is the same record that you're trying to put
into the store.

If you were to call "put" from a punctuator, or do a
`range()` query and then update one of those records with
`put()`, you'd have a very subtle bug on your hands. Right
now, the Streams component that actually calls the Processor
takes care to set the right record context before invoking
the method, and in the case of caching, etc., it also takes
care to swap out the old context and keep it somewhere safe.
But when it comes to public API Processors calling methods
on StateStores, there's no opportunity for any component to
make sure the context is always correct.

In the face of that situation, it seemed better to just move
in the direction of a "normal" data store. I.e., when you
use a HashMap or RocksDB or other "state stores", you don't
expect them to automatically know extra stuff about the
record you're storing. If you need them to know something,
you just put it in the value.

All of that said, I'm just reasoning from first principles
here. To really know if this is a mistake or not, I need to
be in your place. So please push back if you think what I
said is nonsense. My personal plan was to keep an eye out
during the period where the old API was still present, but
deprecated, to see if people were struggling to use the new
API. If so, then we'd have a chance to address it before
dropping the old API. But it's even better if you can help
think it through now.

It did also cross my mind to _not_ add the
StateStoreContext, but just to continue to punt on the
question by just dropping in the new ProcessorContext to the
new init method. If StateStoreContext seems too bold, we can
go that direction. But if we actually add some methods to
StateStoreContext, I'd like to be able to ensure they would
be well defined. I think the current situation was more of
an oversight than a choice.

Thanks again for your reply,
-John


On Wed, 2020-09-09 at 21:23 -0500, Paul Whalen wrote:
> John,
> 
> It's exciting to see this KIP head in this direction!  In the last year or
> so I've tried to sketch out some usability improvements for custom state
> stores, and I also ended up splitting out the StateStoreContext from the
> ProcessorContext in an attempt to facilitate what I was doing.  I sort of
> abandoned it when I realized how large the ideal change might have to be,
> but it's great to see that there is other interest in moving in this
> direction (from the folks that matter :) ).
> 
> Having taken a stab at it myself, I have a comment/question on this bullet
> about StateStoreContext:
> 
> It does *not*  include anything processor- or record- specific, like
> > `forward()` or any information about the "current" record, which is only a
> > well-defined in the context of the Processor. Processors process one record
> > at a time, but state stores may be used to store and fetch many records, so
> > there is no "current record".
> > 
> 
> I totally agree that record-specific or processor-specific context in a
> state store is often not well-defined and it would be good to separate that
> out, but sometimes it (at least record-specific context) is actually
> useful, for example, passing the record's timestamp through to the
> underlying 

Jenkins build is back to normal : Kafka » kafka-trunk-jdk11 #52

[Apache Jenkins Server]

See 

Re: [VOTE] KIP-478 Strongly Typed Streams Processor API

[Paul Whalen]

John,

It's exciting to see this KIP head in this direction!  In the last year or
so I've tried to sketch out some usability improvements for custom state
stores, and I also ended up splitting out the StateStoreContext from the
ProcessorContext in an attempt to facilitate what I was doing.  I sort of
abandoned it when I realized how large the ideal change might have to be,
but it's great to see that there is other interest in moving in this
direction (from the folks that matter :) ).

Having taken a stab at it myself, I have a comment/question on this bullet
about StateStoreContext:

It does *not*  include anything processor- or record- specific, like
> `forward()` or any information about the "current" record, which is only a
> well-defined in the context of the Processor. Processors process one record
> at a time, but state stores may be used to store and fetch many records, so
> there is no "current record".
>

I totally agree that record-specific or processor-specific context in a
state store is often not well-defined and it would be good to separate that
out, but sometimes it (at least record-specific context) is actually
useful, for example, passing the record's timestamp through to the
underlying storage (or changelog topic):
https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStore.java#L121

You could have the writer client of the state store pass this through, but
it would be nice to be able to write state stores where the client did not
have this responsibility.  I'm not sure if the solution is to add some
things back to StateStoreContext, or make yet another context that
represents record-specific context while inside a state store.

Best,
Paul

On Wed, Sep 9, 2020 at 5:43 PM John Roesler  wrote:

> Hello all,
>
> I've been slowly pushing KIP-478 forward over the last year,
> and I'm happy to say that we're making good progress now.
> However, several issues with the original design have come
> to light.
>
> The major changes:
>
> We discovered that the original plan of just adding generic
> parameters to ProcessorContext was too disruptive, so we are
> now adding a new api.ProcessorContext.
>
> That choice forces us to add a new StateStore.init method
> for the new context, but ProcessorContext really isn't ideal
> for state stores to begin with, so I'm proposing a new
> StateStoreContext for this purpose. In a nutshell, there are
> quite a few methods in ProcessorContext that actually should
> never be called from inside a StateStore.
>
> Also, since there is a new ProcessorContext interface, we
> need a new MockProcessorContext implementation in the test-
> utils module.
>
>
>
> The changeset for the KIP document is here:
>
> https://cwiki.apache.org/confluence/pages/diffpagesbyversion.action?pageId=118172121=14=10
>
> And the KIP itself is here:
>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-478+-+Strongly+typed+Processor+API
>
>
> If you have any concerns, please let me know!
>
> Thanks,
> -John
>
>

Re: How do I place a JIRA in status "Patch Available"? (For KAFKA-10473)

[John Roesler]

Hi James,

Good, I’m glad my incredibly vague response was helpful!

If you let me know your wiki user id, I can grant you edit permission. It’s a 
separate account from Jira. 

Thanks,
John

On Wed, Sep 9, 2020, at 20:45, James Cheng wrote:
> Thanks John. That worked.
> 
> I clicked the button that says "Submit Patch", and a dialog box popped 
> up. I didn't fill out anything additional in the dialog, and clicked 
> "Submit Patch" in the dialog.
> 
> The JIRA is now in status "Patch Available"
> 
> I would like to improve the docs at 
> https://cwiki.apache.org/confluence/display/KAFKA/Contributing+Code+Changes 
>  
> to make this step clearer. It looks like I don't have permissions to edit the 
> page.
> 
> Can someone grant me permissions to edit the page? 
> 
> Or, if that is too difficult, can someone edit the page as follows?
> 
> Change
> 
>   7. Change the status of the JIRA to "Patch Available" if it's ready for 
> review.
> to be
> 
>   7. Change the status of the JIRA to "Patch Available" if it's ready 
> for review. Do this by clicking the "Submit Patch" button in JIRA, and 
> then in the resulting dialog, click "Submit Patch".
> 
> -James
> 
> > On Sep 9, 2020, at 6:24 PM, John Roesler  wrote:
> > 
> > Hi James,
> > 
> > I think the button on Jira says “Add Patch” or something confusing like 
> > that. 
> > 
> > Thanks,
> > John
> > 
> > 
> > On Wed, Sep 9, 2020, at 17:34, James Cheng wrote:
> >> I have a JIRA that I am working on, and a pull request available for it.
> >> 
> >> [KAFKA-10473] Website is missing docs on JMX metrics for partition 
> >> size-on-disk (kafka.log:type=Log,name=*)
> >> https://issues.apache.org/jira/browse/KAFKA-10473
> >> https://github.com/apache/kafka/pull/9276
> >> 
> >> The "Contributing Code Changes" instructions say to
> >>7. Change the status of the JIRA to "Patch Available" if it's ready for 
> >> review.
> >> https://cwiki.apache.org/confluence/display/KAFKA/Contributing+Code+Changes
> >> 
> >> How do I do that? 
> >> * The title of my pull request starts with KAFKA-10473, so the JIRA 
> >> does have a link to the pull request
> >> * I *was* able to assign it to myself and then say "Start progress" and 
> >> now the status says "In Progress".
> >> * But I can't find how to set it to "Patch Available". In the JIRA 
> >> website, I can't find a field or menu item that lets me change the 
> >> status to "Patch Available" . 
> >> 
> >> Thanks,
> >> -James
> >> 
> >> 
> >> 
> 
>

Build failed in Jenkins: Kafka » kafka-trunk-jdk8 #50

[Apache Jenkins Server]

See 


Changes:

[github] KAFKA-10436: Implement KIP-478 Topology changes (#9221)

[github] KAFKA-5636: Improve handling of "early" records in sliding windows 
(#9157)

[github] KAFKA-8334 Make sure the thread which tries to complete delayed reque… 
(#8657)


--
[...truncated 6.52 MB...]

org.apache.kafka.streams.TopologyTestDriverTest > shouldInitProcessor[Eos 
enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldThrowForUnknownTopic[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldThrowForUnknownTopic[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPunctuateOnStreamsTime[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPunctuateOnStreamsTime[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldCaptureGlobalTopicNameIfWrittenInto[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldCaptureGlobalTopicNameIfWrittenInto[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldThrowIfInMemoryBuiltInStoreIsAccessedWithUntypedMethod[Eos enabled = 
false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldThrowIfInMemoryBuiltInStoreIsAccessedWithUntypedMethod[Eos enabled = 
false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldProcessFromSourcesThatMatchMultiplePattern[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldProcessFromSourcesThatMatchMultiplePattern[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > shouldPopulateGlobalStore[Eos 
enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > shouldPopulateGlobalStore[Eos 
enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldThrowIfPersistentBuiltInStoreIsAccessedWithUntypedMethod[Eos enabled = 
false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldThrowIfPersistentBuiltInStoreIsAccessedWithUntypedMethod[Eos enabled = 
false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldAllowPrePopulatingStatesStoresWithCachingEnabled[Eos enabled = false] 
STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldAllowPrePopulatingStatesStoresWithCachingEnabled[Eos enabled = false] 
PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldReturnCorrectPersistentStoreTypeOnly[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldReturnCorrectPersistentStoreTypeOnly[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > shouldRespectTaskIdling[Eos 
enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > shouldRespectTaskIdling[Eos 
enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldUseSourceSpecificDeserializers[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldUseSourceSpecificDeserializers[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > shouldReturnAllStores[Eos 
enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > shouldReturnAllStores[Eos 
enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldNotCreateStateDirectoryForStatelessTopology[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldNotCreateStateDirectoryForStatelessTopology[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldApplyGlobalUpdatesCorrectlyInRecursiveTopologies[Eos enabled = false] 
STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldApplyGlobalUpdatesCorrectlyInRecursiveTopologies[Eos enabled = false] 
PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldReturnAllStoresNames[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldReturnAllStoresNames[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPassRecordHeadersIntoSerializersAndDeserializers[Eos enabled = false] 
STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPassRecordHeadersIntoSerializersAndDeserializers[Eos enabled = false] 
PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldProcessConsumerRecordList[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldProcessConsumerRecordList[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldUseSinkSpecificSerializers[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldUseSinkSpecificSerializers[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldFlushStoreForFirstInput[Eos enabled = false] 

Re: How do I place a JIRA in status "Patch Available"? (For KAFKA-10473)

[James Cheng]

Thanks John. That worked.

I clicked the button that says "Submit Patch", and a dialog box popped up. I 
didn't fill out anything additional in the dialog, and clicked "Submit Patch" 
in the dialog.

The JIRA is now in status "Patch Available"

I would like to improve the docs at 
https://cwiki.apache.org/confluence/display/KAFKA/Contributing+Code+Changes 
 
to make this step clearer. It looks like I don't have permissions to edit the 
page.

Can someone grant me permissions to edit the page? 

Or, if that is too difficult, can someone edit the page as follows?

Change

7. Change the status of the JIRA to "Patch Available" if it's ready for 
review.
to be

7. Change the status of the JIRA to "Patch Available" if it's ready for 
review. Do this by clicking the "Submit Patch" button in JIRA, and then in the 
resulting dialog, click "Submit Patch".

-James

> On Sep 9, 2020, at 6:24 PM, John Roesler  wrote:
> 
> Hi James,
> 
> I think the button on Jira says “Add Patch” or something confusing like that. 
> 
> Thanks,
> John
> 
> 
> On Wed, Sep 9, 2020, at 17:34, James Cheng wrote:
>> I have a JIRA that I am working on, and a pull request available for it.
>> 
>> [KAFKA-10473] Website is missing docs on JMX metrics for partition 
>> size-on-disk (kafka.log:type=Log,name=*)
>> https://issues.apache.org/jira/browse/KAFKA-10473
>> https://github.com/apache/kafka/pull/9276
>> 
>> The "Contributing Code Changes" instructions say to
>>  7. Change the status of the JIRA to "Patch Available" if it's ready for 
>> review.
>> https://cwiki.apache.org/confluence/display/KAFKA/Contributing+Code+Changes
>> 
>> How do I do that? 
>> * The title of my pull request starts with KAFKA-10473, so the JIRA 
>> does have a link to the pull request
>> * I *was* able to assign it to myself and then say "Start progress" and 
>> now the status says "In Progress".
>> * But I can't find how to set it to "Patch Available". In the JIRA 
>> website, I can't find a field or menu item that lets me change the 
>> status to "Patch Available" . 
>> 
>> Thanks,
>> -James
>> 
>> 
>> 

[jira] [Resolved] (KAFKA-10170) ReplicaManager should be responsible for checking delayed operations after appending to the log.

[Chia-Ping Tsai (Jira)]

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

Chia-Ping Tsai resolved KAFKA-10170.

Resolution: Duplicate

> ReplicaManager should be responsible for checking delayed operations after 
> appending to the log.
> 
>
> Key: KAFKA-10170
> URL: https://issues.apache.org/jira/browse/KAFKA-10170
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Chia-Ping Tsai
>Assignee: Chia-Ping Tsai
>Priority: Minor
>
> This issue aims to refactor code to simplify the code of checking delayed 
> operations. This issue is inspired by [~hachikuji] 
> (https://github.com/apache/kafka/pull/8657#discussion_r426943627)
> {quote}
> Currently we have a somewhat convoluted model where ReplicaManager creates 
> delayed operations, but we depend on lower level components like Partition to 
> be aware of them and complete them. This breaks encapsulation.
> Not something we should try to complete in this PR, but as an eventual goal, 
> I think we can consider trying to factor delayed operations out of Partition 
> so that they can be managed by ReplicaManager exclusively. If you assume that 
> is the end state, then we could drop completeDelayedRequests and let 
> ReplicaManager always be responsible for checking delayed operations after 
> appending to the log.
> Other than ReplicaManager, the only caller of this method is 
> GroupMetadataManager which uses it during offset expiration. I think the only 
> reason we do this is because we didn't want to waste purgatory space. I don't 
> think that's a good enough reason to go outside the normal flow. It would be 
> simpler to follow the same path. Potentially we could make the callback an 
> Option so that we still have a way to avoid polluting the purgatory.
> {quote}



--
This message was sent by Atlassian Jira
(v8.3.4#803005)

Re: How do I place a JIRA in status "Patch Available"? (For KAFKA-10473)

[John Roesler]

Hi James,

I think the button on Jira says “Add Patch” or something confusing like that. 

Thanks,
John


On Wed, Sep 9, 2020, at 17:34, James Cheng wrote:
> I have a JIRA that I am working on, and a pull request available for it.
> 
> [KAFKA-10473] Website is missing docs on JMX metrics for partition 
> size-on-disk (kafka.log:type=Log,name=*)
> https://issues.apache.org/jira/browse/KAFKA-10473
> https://github.com/apache/kafka/pull/9276
> 
> The "Contributing Code Changes" instructions say to
>   7. Change the status of the JIRA to "Patch Available" if it's ready for 
> review.
> https://cwiki.apache.org/confluence/display/KAFKA/Contributing+Code+Changes
> 
> How do I do that? 
> * The title of my pull request starts with KAFKA-10473, so the JIRA 
> does have a link to the pull request
> * I *was* able to assign it to myself and then say "Start progress" and 
> now the status says "In Progress".
> * But I can't find how to set it to "Patch Available". In the JIRA 
> website, I can't find a field or menu item that lets me change the 
> status to "Patch Available" . 
> 
> Thanks,
> -James
> 
> 
>

Re: [DISCUSS] KIP-649: Dynamic Client Configuration

[Ryan Dielher]

Hi again,

First of all, I apologize for the duplicate email I did not mean to send 2.

> I added the / dynamic / config `supported.configs` to the KIP. This is a map 
> of client software name and version to a list of supported configs registered 
> with that software name and version. 
> 
> e.g.
> {
> 'ClientInformation(softwareName=apache-kafka-java, 
> softwareVersion=x.y.a-SNAPSHOT)': 'acks',
> 'ClientInformation(softwareName=apache-kafka-java, 
> softwareVersion=x.y.b-SNAPSHOT)': 'acks, enable.idempotence'
> }  

There could be an internal topic for this instead of a dynamic config so that 
ad hoc aggregations can be performed on the config registrations. Would it make 
sense to key config registration by ? This would allow 
compatibility information and the current config values for each client that 
registered with a particular  entity to be aggregated and 
returned to the user in the DescribeClientConfigsResponse. Could clients 
register their configs every time that they update their configs? This way a 
log retention time could be set on the internal topic so that only data from 
active clients is kept.

Best,
Ryan

On 2020/09/08 07:10:52, Ryan Dielher  wrote: 
> Hi Jason,
> 
> Thank you again for all of your feedback, it is greatly appreciated.
> Here are some more changes to the KIP: 
> https://cwiki.apache.org/confluence/pages/diffpagesbyversion.action?pageId=158869615=21=17
> 
> > 1. I wonder if we need to bother with `enable.dynamic.config`, especially
> > if the default is going to be true anyway.
> 
> I removed this / static / config in favor of a / dynamic / config 
> `supported.configs`. This config is explained in more detail below.
> 
> > 3. I'm trying to understand the contract between brokers and clients to
> > support dynamic configurations. I imagine that once this is available,
> > users will have a hard time telling which applications support the
> > capability and which do not. Also, we would likely add new dynamic config
> > support over time which would make this even harder since we cannot
> > retroactively change clients to add support for new dynamic configs. I'm
> > wondering if there is anything we can do to make it easier for users to
> > tell which dynamic configs are available for each application.
> 
> 
> I added the / dynamic / config `supported.configs` to the KIP. This is a map 
> of client software name and version to a list of supported configs registered 
> with that software name and version. 
> 
> e.g.
> {
> 'ClientInformation(softwareName=apache-kafka-java, 
> softwareVersion=x.y.a-SNAPSHOT)': 'acks',
> 'ClientInformation(softwareName=apache-kafka-java, 
> softwareVersion=x.y.b-SNAPSHOT)': 'acks, enable.idempotence'
> }  
> 
> The changes propose that `supported.configs` be serialized as a json and 
> stored in an entity config alongside dynamic configs and quotas for an 
> entity. This config is updated for an entity when producers and consumers 
> register the dynamic configurations that they support with the entity. This 
> registration happens every time that they request configs.
> 
> This makes the information refreshable since `supported.configs` can be 
> deleted from an entity. Clients register configs every time they request 
> configs, so the compatibility information will eventually be added again and 
> continuously updated as new clients associate themselves with the entity.
> 
> > 2. Tying dynamic configurations to clientId has some downsides.
> 
> The new protocols in the changes to the KIP are based on the protocols for 
> client quotas. 
> They allow dynamic configs to be tied to a user principal and optionally to a 
> client-id without trying to fit multiple entity names and types into the 
> {Describe, IncrementalAlter}Configs APIs. They also provide a more expressive 
> and extensible interface for dynamic client config entity names and types.
> 
> > It is common for users to use a different clientId for every application in 
> > a
> > consumer group so that it is easier to tie group members back to where
> > the client is running. This makes setting configurations at an application
> > level cumbersome. The alternative is to use the default, but that means
> > hitting /all/ applications which I think is probably not a good idea. A
> > convenient alternative for consumers would be to use group.id, but we don't
> > have anything similar for the producer. I am wondering if we need to give
> > the clients a separate config label of some kind so that there is a
> > convenient way to group configurations. For example `config.group`. Note
> > that this would be another way to opt into dynamic config support.
> 
> Would it be reasonable to treat consumer entities slightly differently than 
> producer entities since they are different types of clients with different 
> needs?
> Since the design pattern of this new protocol is built in part for 
> extensibility of entity names and types, could consumers be associated with 
> an additional entity 

Re: [VOTE] KIP-478 Strongly Typed Streams Processor API

[John Roesler]

Hello all,

I've been slowly pushing KIP-478 forward over the last year,
and I'm happy to say that we're making good progress now.
However, several issues with the original design have come
to light.

The major changes:

We discovered that the original plan of just adding generic
parameters to ProcessorContext was too disruptive, so we are
now adding a new api.ProcessorContext.

That choice forces us to add a new StateStore.init method
for the new context, but ProcessorContext really isn't ideal
for state stores to begin with, so I'm proposing a new
StateStoreContext for this purpose. In a nutshell, there are
quite a few methods in ProcessorContext that actually should
never be called from inside a StateStore.

Also, since there is a new ProcessorContext interface, we
need a new MockProcessorContext implementation in the test-
utils module.



The changeset for the KIP document is here:
https://cwiki.apache.org/confluence/pages/diffpagesbyversion.action?pageId=118172121=14=10

And the KIP itself is here:
https://cwiki.apache.org/confluence/display/KAFKA/KIP-478+-+Strongly+typed+Processor+API


If you have any concerns, please let me know!

Thanks,
-John

How do I place a JIRA in status "Patch Available"? (For KAFKA-10473)

[James Cheng]

I have a JIRA that I am working on, and a pull request available for it.

[KAFKA-10473] Website is missing docs on JMX metrics for partition size-on-disk 
(kafka.log:type=Log,name=*)
https://issues.apache.org/jira/browse/KAFKA-10473
https://github.com/apache/kafka/pull/9276

The "Contributing Code Changes" instructions say to
7. Change the status of the JIRA to "Patch Available" if it's ready for 
review.
https://cwiki.apache.org/confluence/display/KAFKA/Contributing+Code+Changes

How do I do that? 
* The title of my pull request starts with KAFKA-10473, so the JIRA does have a 
link to the pull request
* I *was* able to assign it to myself and then say "Start progress" and now the 
status says "In Progress".
* But I can't find how to set it to "Patch Available". In the JIRA website, I 
can't find a field or menu item that lets me change the status to "Patch 
Available" . 

Thanks,
-James

[jira] [Created] (KAFKA-10473) Website is missing docs on JMX metrics for partition size-on-disk

[James Cheng (Jira)]

James Cheng created KAFKA-10473:
---

 Summary: Website is missing docs on JMX metrics for partition 
size-on-disk
 Key: KAFKA-10473
 URL: https://issues.apache.org/jira/browse/KAFKA-10473
 Project: Kafka
  Issue Type: Improvement
  Components: docs, documentation
Affects Versions: 2.5.1, 2.6.0
Reporter: James Cheng


The website is missing docs on the following JMX metrics:

kafka.log,type=Log,name=Size

kafka.log,type=Log,name=NumLogSegments

kafka.log,type=Log,name=LogStartOffset

kafka.log,type=Log,name=LogEndOffset

 



--
This message was sent by Atlassian Jira
(v8.3.4#803005)

Re: [VOTE] KIP-654 Aborting Transaction with non-flushed data should throw a non-fatal Exception

[Sophie Blee-Goldman]

+1 from me as well (non-binding)

Gokul, it looks you've now received enough binding votes and the vote has
been
open for sufficiently long. You can conclude the vote and open the PR for
review
when it's ready

Cheers,
Sophie

On Wed, Sep 9, 2020 at 2:47 PM Guozhang Wang  wrote:

> +1. Thanks for the KIP Gokul !
>
> Guozhang
>
> On Tue, Sep 8, 2020 at 6:52 PM Matthias J. Sax  wrote:
>
> > +1 (binding)
> >
> > On 9/8/20 2:49 PM, Jason Gustafson wrote:
> > > +1 Thanks for the KIP!
> > >
> > > On Thu, Sep 3, 2020 at 12:25 PM Gokul Srinivas 
> > wrote:
> > >
> > >> Hi,
> > >>
> > >> I would like to call a vote on the following KIP.
> > >>
> > >> *KIP *-
> > >>
> > >>
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-654:+Aborted+transaction+with+non-flushed+data+should+throw+a+non-fatal+exception
> > >> <
> > >>
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-654:+Aborted+transaction+with+non-flushed+data+should+throw+a+non-fatal+exception
> > >>>
> > >>
> > >> TL;DR: This KIP proposes to throw a new, non-fatal exception whilst
> > >> aborting transactions with non-flushed data. This will help users
> > >> distinguish non-fatal errors and potentially retry the batch.
> > >>
> > >> Thanks,
> > >> -Gokul
> > >>
> > >>
> > >
> >
> >
>
> --
> -- Guozhang
>

Re: [VOTE] KIP-654 Aborting Transaction with non-flushed data should throw a non-fatal Exception

[Guozhang Wang]

+1. Thanks for the KIP Gokul !

Guozhang

On Tue, Sep 8, 2020 at 6:52 PM Matthias J. Sax  wrote:

> +1 (binding)
>
> On 9/8/20 2:49 PM, Jason Gustafson wrote:
> > +1 Thanks for the KIP!
> >
> > On Thu, Sep 3, 2020 at 12:25 PM Gokul Srinivas 
> wrote:
> >
> >> Hi,
> >>
> >> I would like to call a vote on the following KIP.
> >>
> >> *KIP *-
> >>
> >>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-654:+Aborted+transaction+with+non-flushed+data+should+throw+a+non-fatal+exception
> >> <
> >>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-654:+Aborted+transaction+with+non-flushed+data+should+throw+a+non-fatal+exception
> >>>
> >>
> >> TL;DR: This KIP proposes to throw a new, non-fatal exception whilst
> >> aborting transactions with non-flushed data. This will help users
> >> distinguish non-fatal errors and potentially retry the batch.
> >>
> >> Thanks,
> >> -Gokul
> >>
> >>
> >
>
>

-- 
-- Guozhang

Re: [DISCUSS] KIP-663: API to Start and Shut Down Stream Threads and to Request Closing of Kafka Streams Clients

[Guozhang Wang]

Hello Bruno,

Finally got some time to review your KIP and the discussion thread now.. a
few comments below:

1) I'm with Matthias about the newly added numberOfAliveStreamThreads v.s.
existing localThreadsMetadata: to me it seems we can always achieve the
first based on the second. It seems not worthy to provide some "syntax
sugar" to the API but just let users do the filtering themselves.
Furthermore, I'm wondering what's the rationale behind removing the DEAD
threads from localThreadsMetadata()? Personally I feel returning all
threads, including those who are ever closed, either due to exception or
due to removeStreamThread, would be beneficial for debugging purposes, as
long as within the lifetime of an instance we expect the amount of such
dead threads will not increase linearly --- and if we agree with that,
maybe we can rename "removeStreamThread" to sth. like
"terminateStreamThread" indicating it is only terminated but not removed
--- and of course if users do not want to see those DEAD threads they can
always filter them out. I'm just proposing that we should still leave the
door open for those who want to check those ever terminated threads.

2) I think it would help to write down some example user code in exception
handler e.g. to illustrate how this would be implemented -- e.g. we know
that practically the handler need to maintain a "this" reference of the
instance anyways in order to shutdown the whole instance or, add/terminate
threads dynamically, but I want to see if we have listed all possible call
paths like: a) a thread's handler logic to terminate another thread, b) a
thread handler to add new threads, etc are all appropriately supported
without deadlocks.


Guozhang


On Wed, Sep 9, 2020 at 11:35 AM Matthias J. Sax  wrote:

> I would prefer to not add a new method. It seems unnecessary.
> `localThreadMetadata` does return all threads in all states(*) and thus
> provides full insight.
>
> (*) A thread in state DEAD could be returned as long as it's not removed
> yet.
>
> I don't see any advantage to pre-filter threads and to exclude threads
> in state CREATE or PENDING_SHUTDOWN. Even if a CREATED thread is not
> started yet, it is still "alive" in a broader sense. For example, if a
> user wants to scale out to 10 thread, and 8 are RUNNING and 2 are in
> state CREATED, a user won't need to add 2 more threads -- there are
> already 10 threads.
>
> For PENDING_SHUTDOWN and scale in it would be different I guess, as the
> proposal would be to filter them out right away. However, filtering them
> seems actually not to be "correct", as a thread in PENDING_SHUTDOWN
> might still process data and it's thus still "alive".
>
> If there is still a need later to add a new method about "alive thread"
> we can always add as a follow up -- removing things is much harder.
>
> I also don't think that there is value in returning names of dead
> threads, as we recycle names.
>
>
> -Matthias
>
>
> On 9/9/20 10:04 AM, Sophie Blee-Goldman wrote:
> > I agree that the current behavior of localThreadsMetadata() does not seem
> > to match, but it seems like we will be forced to change it to only return
> > currently-alive threads. For one thing, we plan to recycle old thread
> names.
> > It would be pretty confusing for a user to get two (or more)
> ThreadMetadata
> > objects returned with the same name, since AFAICT this is the only
> > distinguishing identifier of stream threads. I think we should enforce
> that
> > only live threads are returned by localThreadsMetadata(). Plus, as
> Matthias
> > pointed out, we plan to remove dead threads from the KafkaStreams client,
> > so still returning them in the metadata would be extremely odd.
> >
> > If we think that there might be some use case that requires knowing which
> > threads have died, we could consider adding a method that returns the
> > names of dead threads. But the only use case I can imagine would probably
> > be better served by a callback that gets invoked when the thread dies,
> which
> > we already have.
> >
> > On Tue, Sep 8, 2020 at 11:46 PM Bruno Cadonna 
> wrote:
> >
> >> Hi Matthias and Sophie,
> >>
> >> I agree that localThreadsMetadata() can be used here. However,
> >> localThreadsMetadata() returns all stream threads irrespectively of
> >> their states. Alive stream threads are specified as being in one of the
> >> following states: RUNNING, STARTING, PARTITIONS_REVOKED, and
> >> PARTITIONS_ASSIGNED. Hence, users would need to filter the result of
> >> localThreadsMetadata(). I thought, it would be neat to have a method
> >> that hides this filtering and returns the number of alive stream
> >> threads, because that is the most basic information you might need to
> >> decide about adding or removing stream threads. For all more advanced
> >> use cases users should use localThreadsMetadata(). I am also happy with
> >> removing the method. WDYT?
> >>
> >> Best,
> >> Bruno
> >>
> >> On 09.09.20 03:51, Matthias J. Sax wrote:
> >>> Currently 

Build failed in Jenkins: Kafka » kafka-trunk-jdk11 #51

[Apache Jenkins Server]

See 


Changes:

[github] KAFKA-10436: Implement KIP-478 Topology changes (#9221)

[github] KAFKA-5636: Improve handling of "early" records in sliding windows 
(#9157)


--
[...truncated 3.28 MB...]

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfTimestampIsDifferentForCompareValueTimestamp PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfKeyAndValueAndTimestampIsEqualWithNullForCompareKeyValueTimestampWithProducerRecord
 STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfKeyAndValueAndTimestampIsEqualWithNullForCompareKeyValueTimestampWithProducerRecord
 PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentWithNullForCompareKeyValueWithProducerRecord STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentWithNullForCompareKeyValueWithProducerRecord PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentForCompareKeyValueTimestampWithProducerRecord 
STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentForCompareKeyValueTimestampWithProducerRecord PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldNotAllowNullProducerRecordWithExpectedRecordForCompareValueTimestamp 
STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldNotAllowNullProducerRecordWithExpectedRecordForCompareValueTimestamp 
PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldNotAllowNullExpectedRecordForCompareValue STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldNotAllowNullExpectedRecordForCompareValue PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentWithNullReversForCompareKeyValueTimestampWithProducerRecord
 STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentWithNullReversForCompareKeyValueTimestampWithProducerRecord
 PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldNotAllowNullProducerRecordForCompareKeyValue STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldNotAllowNullProducerRecordForCompareKeyValue PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentWithNullReversForCompareKeyValueWithProducerRecord 
STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentWithNullReversForCompareKeyValueWithProducerRecord 
PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfKeyAndValueAndTimestampIsEqualForCompareKeyValueTimestamp STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfKeyAndValueAndTimestampIsEqualForCompareKeyValueTimestamp PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentWithNullForCompareKeyValueTimestampWithProducerRecord
 STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentWithNullForCompareKeyValueTimestampWithProducerRecord
 PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfKeyIsDifferentWithNullReversForCompareKeyValueWithProducerRecord 
STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfKeyIsDifferentWithNullReversForCompareKeyValueWithProducerRecord 
PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfKeyAndValueIsEqualWithNullForCompareKeyValueWithProducerRecord 
STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfKeyAndValueIsEqualWithNullForCompareKeyValueWithProducerRecord 
PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfKeyIsDifferentWithNullForCompareKeyValueTimestampWithProducerRecord 
STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfKeyIsDifferentWithNullForCompareKeyValueTimestampWithProducerRecord 
PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentForCompareKeyValueTimestamp STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentForCompareKeyValueTimestamp PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfValueAndTimestampIsEqualForCompareValueTimestampWithProducerRecord 
STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfValueAndTimestampIsEqualForCompareValueTimestampWithProducerRecord 
PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentWithNullReversForCompareKeyValueTimestamp STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentWithNullReversForCompareKeyValueTimestamp PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentWithNullReverseForCompareValueTimestampWithProducerRecord
 STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 

Re: [DISCUSS] Apache Kafka 2.7.0 release

[Bill Bejeck]

Hi Dongjin,

I've moved both KIPs to the release plan.

Keep in mind the cutoff for KIP acceptance is September 30th. If the KIP
discussions are completed, I'd recommend starting a vote for them.

Regards,
Bill

On Wed, Sep 9, 2020 at 8:39 AM Dongjin Lee  wrote:

> Hi Bill,
>
> Could you add the following KIPs to the plan?
>
> - KIP-508: Make Suppression State Queriable
> <
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-508%3A+Make+Suppression+State+Queriable
> >
> - KIP-653: Upgrade log4j to log4j2
> <
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-653%3A+Upgrade+log4j+to+log4j2
> >
>
> Both KIPs are completely implemented with passing all tests, but not got
> reviewed by the committers. Could anyone have a look?
>
> Thanks,
> Dongjin
>
> On Wed, Sep 9, 2020 at 8:38 AM Leah Thomas  wrote:
>
> > Hi Bill,
> >
> > Could you also add KIP-450 to the release plan? It's been merged.
> >
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-450%3A+Sliding+Window+Aggregations+in+the+DSL
> >
> > Cheers,
> > Leah
> >
> > On Tue, Sep 8, 2020 at 9:32 AM Bill Bejeck  wrote:
> >
> > > Hi Bruno,
> > >
> > > Thanks for letting me know, I've added KIP-662 to the release plan.
> > >
> > > -Bill
> > >
> > > On Mon, Sep 7, 2020 at 11:33 AM Bruno Cadonna 
> > wrote:
> > >
> > > > Hi Bill,
> > > >
> > > > Could you add KIP-662 [1] to the release plan. The KIP has been
> already
> > > > implemented.
> > > >
> > > > Best,
> > > > Bruno
> > > >
> > > >
> > > > [1]
> > > >
> > > >
> > >
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-662%3A+Throw+Exception+when+Source+Topics+of+a+Streams+App+are+Deleted
> > > >
> > > > On 26.08.20 16:54, Bill Bejeck wrote:
> > > > > Greetings All!
> > > > >
> > > > > I've published a release plan at
> > > > >
> > > >
> > >
> >
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=158872629
> > > > .
> > > > > I have included all of the KIPs that are currently approved, but
> I'm
> > > > happy
> > > > > to make any adjustments as necessary.
> > > > >
> > > > > The KIP freeze is on September 30 with a target release date of
> > > November
> > > > 6.
> > > > >
> > > > > Let me know if there are any objections.
> > > > >
> > > > > Thanks,
> > > > > Bill Bejeck
> > > > >
> > > > > On Fri, Aug 14, 2020 at 4:01 PM John Roesler 
> > > > wrote:
> > > > >
> > > > >> Thanks, Bill!
> > > > >> -John
> > > > >>
> > > > >> On Thu, 2020-08-13 at 15:19 -0700, Ismael Juma wrote:
> > > > >>> Thanks for volunteering Bill. :)
> > > > >>>
> > > > >>> Ismael
> > > > >>>
> > > > >>> On Thu, Aug 13, 2020 at 3:13 PM Bill Bejeck 
> > > > wrote:
> > > > >>>
> > > >  Hi All,
> > > > 
> > > >  I'd like to volunteer to be the release manager for our next
> > feature
> > > >  release, 2.7. If there are no objections, I'll send out the
> > release
> > > > >> plan
> > > >  soon.
> > > > 
> > > >  Thanks,
> > > >  Bill Bejeck
> > > > 
> > > > >>
> > > > >>
> > > > >
> > > >
> > >
> >
>
>
> --
> *Dongjin Lee*
>
> *A hitchhiker in the mathematical world.*
>
>
>
>
> *github:  github.com/dongjinleekr
> keybase: https://keybase.io/dongjinleekr
> linkedin: kr.linkedin.com/in/dongjinleekr
> speakerdeck:
> speakerdeck.com/dongjin
> *
>

Re: [DISCUSS] Apache Kafka 2.7.0 release

[Bill Bejeck]

Hi Leah,

Thanks for the heads up, I've added KIP-450 to the release plan.

-Bill

On Tue, Sep 8, 2020 at 7:38 PM Leah Thomas  wrote:

> Hi Bill,
>
> Could you also add KIP-450 to the release plan? It's been merged.
>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-450%3A+Sliding+Window+Aggregations+in+the+DSL
>
> Cheers,
> Leah
>
> On Tue, Sep 8, 2020 at 9:32 AM Bill Bejeck  wrote:
>
> > Hi Bruno,
> >
> > Thanks for letting me know, I've added KIP-662 to the release plan.
> >
> > -Bill
> >
> > On Mon, Sep 7, 2020 at 11:33 AM Bruno Cadonna 
> wrote:
> >
> > > Hi Bill,
> > >
> > > Could you add KIP-662 [1] to the release plan. The KIP has been already
> > > implemented.
> > >
> > > Best,
> > > Bruno
> > >
> > >
> > > [1]
> > >
> > >
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-662%3A+Throw+Exception+when+Source+Topics+of+a+Streams+App+are+Deleted
> > >
> > > On 26.08.20 16:54, Bill Bejeck wrote:
> > > > Greetings All!
> > > >
> > > > I've published a release plan at
> > > >
> > >
> >
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=158872629
> > > .
> > > > I have included all of the KIPs that are currently approved, but I'm
> > > happy
> > > > to make any adjustments as necessary.
> > > >
> > > > The KIP freeze is on September 30 with a target release date of
> > November
> > > 6.
> > > >
> > > > Let me know if there are any objections.
> > > >
> > > > Thanks,
> > > > Bill Bejeck
> > > >
> > > > On Fri, Aug 14, 2020 at 4:01 PM John Roesler 
> > > wrote:
> > > >
> > > >> Thanks, Bill!
> > > >> -John
> > > >>
> > > >> On Thu, 2020-08-13 at 15:19 -0700, Ismael Juma wrote:
> > > >>> Thanks for volunteering Bill. :)
> > > >>>
> > > >>> Ismael
> > > >>>
> > > >>> On Thu, Aug 13, 2020 at 3:13 PM Bill Bejeck 
> > > wrote:
> > > >>>
> > >  Hi All,
> > > 
> > >  I'd like to volunteer to be the release manager for our next
> feature
> > >  release, 2.7. If there are no objections, I'll send out the
> release
> > > >> plan
> > >  soon.
> > > 
> > >  Thanks,
> > >  Bill Bejeck
> > > 
> > > >>
> > > >>
> > > >
> > >
> >
>

Jenkins build is back to normal : Kafka » kafka-trunk-jdk8 #49

[Apache Jenkins Server]

See 

[GitHub] [kafka-site] vvcephei merged pull request #301: Redo 2.6 docs commit

[GitBox]

vvcephei merged pull request #301:
URL: https://github.com/apache/kafka-site/pull/301


   



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

Jenkins build is back to normal : Kafka » kafka-trunk-jdk15 #50

[Apache Jenkins Server]

See 

Re: [DISCUSS] KIP-663: API to Start and Shut Down Stream Threads and to Request Closing of Kafka Streams Clients

[Matthias J. Sax]

I would prefer to not add a new method. It seems unnecessary.
`localThreadMetadata` does return all threads in all states(*) and thus
provides full insight.

(*) A thread in state DEAD could be returned as long as it's not removed
yet.

I don't see any advantage to pre-filter threads and to exclude threads
in state CREATE or PENDING_SHUTDOWN. Even if a CREATED thread is not
started yet, it is still "alive" in a broader sense. For example, if a
user wants to scale out to 10 thread, and 8 are RUNNING and 2 are in
state CREATED, a user won't need to add 2 more threads -- there are
already 10 threads.

For PENDING_SHUTDOWN and scale in it would be different I guess, as the
proposal would be to filter them out right away. However, filtering them
seems actually not to be "correct", as a thread in PENDING_SHUTDOWN
might still process data and it's thus still "alive".

If there is still a need later to add a new method about "alive thread"
we can always add as a follow up -- removing things is much harder.

I also don't think that there is value in returning names of dead
threads, as we recycle names.


-Matthias


On 9/9/20 10:04 AM, Sophie Blee-Goldman wrote:
> I agree that the current behavior of localThreadsMetadata() does not seem
> to match, but it seems like we will be forced to change it to only return
> currently-alive threads. For one thing, we plan to recycle old thread names.
> It would be pretty confusing for a user to get two (or more) ThreadMetadata
> objects returned with the same name, since AFAICT this is the only
> distinguishing identifier of stream threads. I think we should enforce that
> only live threads are returned by localThreadsMetadata(). Plus, as Matthias
> pointed out, we plan to remove dead threads from the KafkaStreams client,
> so still returning them in the metadata would be extremely odd.
> 
> If we think that there might be some use case that requires knowing which
> threads have died, we could consider adding a method that returns the
> names of dead threads. But the only use case I can imagine would probably
> be better served by a callback that gets invoked when the thread dies, which
> we already have.
> 
> On Tue, Sep 8, 2020 at 11:46 PM Bruno Cadonna  wrote:
> 
>> Hi Matthias and Sophie,
>>
>> I agree that localThreadsMetadata() can be used here. However,
>> localThreadsMetadata() returns all stream threads irrespectively of
>> their states. Alive stream threads are specified as being in one of the
>> following states: RUNNING, STARTING, PARTITIONS_REVOKED, and
>> PARTITIONS_ASSIGNED. Hence, users would need to filter the result of
>> localThreadsMetadata(). I thought, it would be neat to have a method
>> that hides this filtering and returns the number of alive stream
>> threads, because that is the most basic information you might need to
>> decide about adding or removing stream threads. For all more advanced
>> use cases users should use localThreadsMetadata(). I am also happy with
>> removing the method. WDYT?
>>
>> Best,
>> Bruno
>>
>> On 09.09.20 03:51, Matthias J. Sax wrote:
>>> Currently we, don't cleanup dead threads, but the KIP proposes to change
>>> this:
>>>
 Stream threads that are in state DEAD will be removed from the stream
>> threads of a Kafka Streams client.
>>>
>>>
>>> -Matthias
>>>
>>> On 9/8/20 2:37 PM, Sophie Blee-Goldman wrote:
 Ah, I forgot about localThreadsMetadata(). In that. case I agree,
>> there's
 no reason
 to introduce a new method when we can get both the names and number of
>> all
 running threads from this.

 I assume that we would update localThreadsMetadata to only return
>> currently
 alive threads as part of this KIP -- at a quick glance, it seems like we
 don't do
 any pruning of dead threads at the moment

 On Tue, Sep 8, 2020 at 1:58 PM Matthias J. Sax 
>> wrote:

> I am not sure if we need a new method? There is already
> `localThreadsMetadata()`. What do we gain by adding a new one?
>
> Returning the thread's name (as `Optional`) for both add() and
> remove() is fine with me.
>
>
> -Matthias
>
> On 9/8/20 12:58 PM, Sophie Blee-Goldman wrote:
>> Sorry Bruno, I think I missed the end of your message with the
>> numberOfAliveStreamThreads()
>> proposal. I agree, that would be better than the alternatives I
>> listed.
>> That said:
>>
>>> They rather suggest that the method returns a list of handles to the
>> stream threads.
>>
>> I hadn't thought of that originally, but now that you mention it, this
>> might be a good idea.
>> I don't think we should return actual handles on the threads, but
>> maybe a
>> list of the thread
>> names rather than a single number of currently alive threads.
>>
>> Since we seem to think it would be difficult if not impossible to keep
>> track of the number
>> of running stream threads, we should apply the same reasoning to the
> names

Build failed in Jenkins: Kafka » kafka-trunk-jdk11 #50

[Apache Jenkins Server]

See 


Changes:

[github] KAFKA-10403: Replace Scala collection by Java collection in 
Log4jController (#9182)

[github] MINOR: mirror integration tests should not call System.exit (#9200)


--
[...truncated 3.29 MB...]
org.apache.kafka.streams.TopologyTestDriverTest > 
shouldCaptureGlobalTopicNameIfWrittenInto[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldThrowIfInMemoryBuiltInStoreIsAccessedWithUntypedMethod[Eos enabled = 
false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldThrowIfInMemoryBuiltInStoreIsAccessedWithUntypedMethod[Eos enabled = 
false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldProcessFromSourcesThatMatchMultiplePattern[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldProcessFromSourcesThatMatchMultiplePattern[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > shouldPopulateGlobalStore[Eos 
enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > shouldPopulateGlobalStore[Eos 
enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldThrowIfPersistentBuiltInStoreIsAccessedWithUntypedMethod[Eos enabled = 
false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldThrowIfPersistentBuiltInStoreIsAccessedWithUntypedMethod[Eos enabled = 
false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldAllowPrePopulatingStatesStoresWithCachingEnabled[Eos enabled = false] 
STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldAllowPrePopulatingStatesStoresWithCachingEnabled[Eos enabled = false] 
PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldReturnCorrectPersistentStoreTypeOnly[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldReturnCorrectPersistentStoreTypeOnly[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > shouldRespectTaskIdling[Eos 
enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > shouldRespectTaskIdling[Eos 
enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldUseSourceSpecificDeserializers[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldUseSourceSpecificDeserializers[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > shouldReturnAllStores[Eos 
enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > shouldReturnAllStores[Eos 
enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldNotCreateStateDirectoryForStatelessTopology[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldNotCreateStateDirectoryForStatelessTopology[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldApplyGlobalUpdatesCorrectlyInRecursiveTopologies[Eos enabled = false] 
STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldApplyGlobalUpdatesCorrectlyInRecursiveTopologies[Eos enabled = false] 
PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldReturnAllStoresNames[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldReturnAllStoresNames[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPassRecordHeadersIntoSerializersAndDeserializers[Eos enabled = false] 
STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPassRecordHeadersIntoSerializersAndDeserializers[Eos enabled = false] 
PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldProcessConsumerRecordList[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldProcessConsumerRecordList[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldUseSinkSpecificSerializers[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldUseSinkSpecificSerializers[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldFlushStoreForFirstInput[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldFlushStoreForFirstInput[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldProcessFromSourceThatMatchPattern[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldProcessFromSourceThatMatchPattern[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldCaptureSinkTopicNamesIfWrittenInto[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldCaptureSinkTopicNamesIfWrittenInto[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldUpdateStoreForNewKey[Eos enabled = false] STARTED

Re: [DISCUSS] KIP-663: API to Start and Shut Down Stream Threads and to Request Closing of Kafka Streams Clients

[Sophie Blee-Goldman]

I agree that the current behavior of localThreadsMetadata() does not seem
to match, but it seems like we will be forced to change it to only return
currently-alive threads. For one thing, we plan to recycle old thread names.
It would be pretty confusing for a user to get two (or more) ThreadMetadata
objects returned with the same name, since AFAICT this is the only
distinguishing identifier of stream threads. I think we should enforce that
only live threads are returned by localThreadsMetadata(). Plus, as Matthias
pointed out, we plan to remove dead threads from the KafkaStreams client,
so still returning them in the metadata would be extremely odd.

If we think that there might be some use case that requires knowing which
threads have died, we could consider adding a method that returns the
names of dead threads. But the only use case I can imagine would probably
be better served by a callback that gets invoked when the thread dies, which
we already have.

On Tue, Sep 8, 2020 at 11:46 PM Bruno Cadonna  wrote:

> Hi Matthias and Sophie,
>
> I agree that localThreadsMetadata() can be used here. However,
> localThreadsMetadata() returns all stream threads irrespectively of
> their states. Alive stream threads are specified as being in one of the
> following states: RUNNING, STARTING, PARTITIONS_REVOKED, and
> PARTITIONS_ASSIGNED. Hence, users would need to filter the result of
> localThreadsMetadata(). I thought, it would be neat to have a method
> that hides this filtering and returns the number of alive stream
> threads, because that is the most basic information you might need to
> decide about adding or removing stream threads. For all more advanced
> use cases users should use localThreadsMetadata(). I am also happy with
> removing the method. WDYT?
>
> Best,
> Bruno
>
> On 09.09.20 03:51, Matthias J. Sax wrote:
> > Currently we, don't cleanup dead threads, but the KIP proposes to change
> > this:
> >
> >> Stream threads that are in state DEAD will be removed from the stream
> threads of a Kafka Streams client.
> >
> >
> > -Matthias
> >
> > On 9/8/20 2:37 PM, Sophie Blee-Goldman wrote:
> >> Ah, I forgot about localThreadsMetadata(). In that. case I agree,
> there's
> >> no reason
> >> to introduce a new method when we can get both the names and number of
> all
> >> running threads from this.
> >>
> >> I assume that we would update localThreadsMetadata to only return
> currently
> >> alive threads as part of this KIP -- at a quick glance, it seems like we
> >> don't do
> >> any pruning of dead threads at the moment
> >>
> >> On Tue, Sep 8, 2020 at 1:58 PM Matthias J. Sax 
> wrote:
> >>
> >>> I am not sure if we need a new method? There is already
> >>> `localThreadsMetadata()`. What do we gain by adding a new one?
> >>>
> >>> Returning the thread's name (as `Optional`) for both add() and
> >>> remove() is fine with me.
> >>>
> >>>
> >>> -Matthias
> >>>
> >>> On 9/8/20 12:58 PM, Sophie Blee-Goldman wrote:
>  Sorry Bruno, I think I missed the end of your message with the
>  numberOfAliveStreamThreads()
>  proposal. I agree, that would be better than the alternatives I
> listed.
>  That said:
> 
> > They rather suggest that the method returns a list of handles to the
>  stream threads.
> 
>  I hadn't thought of that originally, but now that you mention it, this
>  might be a good idea.
>  I don't think we should return actual handles on the threads, but
> maybe a
>  list of the thread
>  names rather than a single number of currently alive threads.
> 
>  Since we seem to think it would be difficult if not impossible to keep
>  track of the number
>  of running stream threads, we should apply the same reasoning to the
> >>> names
>  and not
>  assume the user can/will keep track of every thread returned by
>  addStreamThread() or
>  removeStreamThread(). Users should generally take any required action
>  immediately
>  after adding/removing the thread -- eg deregistering the thread
> metrics
> >>> --
>  but it might
>  still be useful to provide a convenience method listing all of the
> >>> current
>  threads
> 
>  And of course you could still get the number of threads easily by
> >>> invoking
>  size() on the
>  returned list (or ordered set?).
> 
>  On Tue, Sep 8, 2020 at 12:16 PM Bruno Cadonna 
> >>> wrote:
> 
> > Thank you again for the feedback Sophie!
> >
> > As I tried to point out in my previous e-mail, removing a stream
> thread
> > from a Kafka Streams client that does not have alive stream threads
> is
> > nothing exceptional for the client per se. However, it can become
> > exceptional within the context of the user. For example, if users
> want
> > to remove a stream thread from a client without alive stream threads
> > because one if their metrics say so, then this is exceptional in the
> > context of that user metric not in 

Re: [DISCUSS] KIP-516: Topic Identifiers

[Justine Olshan]

Hello all, it's been almost a year! I've made some changes to this KIP and hope 
to continue the discussion. 

One of the main changes I've added is now the metadata response will include 
the topic ID (as Colin suggested). Clients can obtain the topicID of a given 
topic through a TopicDescription. The topicId will also be included with the 
UpdateMetadata request. 

Let me know what you all think.
Thank you,
Justine

On 2019/09/13 16:38:26, "Colin McCabe"  wrote: 
> Hi Lucas,
> 
> Thanks for tackling this.  Topic IDs are a great idea, and this is a really 
> good writeup.
> 
> For /brokers/topics/[topic], the schema version should be bumped to version 
> 3, rather than 2.  KIP-455 bumped the version of this znode to 2 already :)
> 
> Given that we're going to be seeing these things as strings as lot (in logs, 
> in ZooKeeper, on the command-line, etc.), does it make sense to use base64 
> when converting them to strings?
> 
> Here is an example of the hex representation:
> 6fcb514b-b878-4c9d-95b7-8dc3a7ce6fd8
> 
> And here is an example in base64.
> b8tRS7h4TJ2Vt43Dp85v2A
> 
> The base64 version saves 15 letters (to be fair, 4 of those were dashes that 
> we could have elided in the hex representation.)
> 
> Another thing to consider is that we should specify that the all-zeroes UUID 
> is not a valid topic UUID.   We can't use null for this because we can't pass 
> a null UUID over the RPC protocol (there is no special pattern for null, nor 
> do we want to waste space reserving such a pattern.)
> 
> Maybe I missed it, but did you describe "migration of... existing topic[s] 
> without topic IDs" in detail in any section?  It seems like when the new 
> controller becomes active, it should just generate random UUIDs for these, 
> and write the random UUIDs back to ZooKeeper.  It would be good to spell that 
> out.  We should make it clear that this happens regardless of the 
> inter-broker protocol version (it's a compatible change).
> 
> "LeaderAndIsrRequests including an is_every_partition flag" seems a bit 
> wordy.  Can we just call these "full LeaderAndIsrRequests"?  Then the RPC 
> field could be named "full".  Also, it would probably be better for the RPC 
> field to be an enum of { UNSPECIFIED, INCREMENTAL, FULL }, so that we can 
> cleanly handle old versions (by treating them as UNSPECIFIED)
> 
> In the LeaderAndIsrRequest section, you write "A final deletion event will be 
> secheduled for X ms after the LeaderAndIsrRequest was first received..."  I 
> guess the X was a placeholder that you intended to replace before posting? :) 
>  In any case, this seems like the kind of thing we'd want a configuration 
> for.  Let's describe that configuration key somewhere in this KIP, including 
> what its default value is.
> 
> We should probably also log a bunch of messages at WARN level when something 
> is scheduled for deletion, as well.  (Maybe this was assumed, but it would be 
> good to mention it).
> 
> I feel like there are a few sections that should be moved to "rejected 
> alternatives."  For example, in the DeleteTopics section, since we're not 
> going to do option 1 or 2, these should be moved into "rejected 
> alternatives,"  rather than appearing inline.  Another case is the "Should we 
> remove topic name from the protocol where possible" section.  This is clearly 
> discussing a design alternative that we're not proposing to implement: 
> removing the topic name from those protocols.
> 
> Is it really necessary to have a new /admin/delete_topics_by_id path in 
> ZooKeeper?  It seems like we don't really need this.  Whenever there is a new 
> controller, we'll send out full LeaderAndIsrRequests which will trigger the 
> stale topics to be cleaned up.   The active controller will also send the 
> full LeaderAndIsrRequest to brokers that are just starting up.So we don't 
> really need this kind of two-phase commit (send out StopReplicasRequest, get 
> ACKs from all nodes, commit by removing /admin/delete_topics node) any more.
> 
> You mention that FetchRequest will now include UUID to avoid issues where 
> requests are made to stale partitions.  However, adding a UUID to 
> MetadataRequest is listed as future work, out of scope for this KIP.  How 
> will the client learn what the topic UUID is, if the metadata response 
> doesn't include that information?  It seems like adding the UUID to 
> MetadataResponse would be an improvement here that might not be too hard to 
> make.
> 
> best,
> Colin
> 
> 
> On Mon, Sep 9, 2019, at 17:48, Ryanne Dolan wrote:
> > Lucas, this would be great. I've run into issues with topics being
> > resurrected accidentally, since a client cannot easily distinguish between
> > a deleted topic and a new topic with the same name. I'd need the ID
> > accessible from the client to solve that issue, but this is a good first
> > step.
> > 
> > Ryanne
> > 
> > On Wed, Sep 4, 2019 at 1:41 PM Lucas Bradstreet  wrote:
> > 
> > > Hi all,
> > >
> > > I would like to kick 

Re: [DISCUSS] KIP-664: Provide tooling to detect and abort hanging transactions

[Jason Gustafson]

Hey Tom,

Yeah, that's fair. I will update the proposal. I was also thinking of
adding a separate column for duration, just to save users the trouble of
computing it.

Thanks,
Jason

On Wed, Sep 9, 2020 at 1:21 AM Tom Bentley  wrote:

> Hi Jason,
>
> The KIP looks good to me, but I had one question. AFAIU the LastTimestamp
> column in the output of --describe-producers and --find-hanging is there so
> the users of the tool know the txnLastUpdateTimestamp of the
> TransactionMetadata and from that and the (max) timeout can infer something
> about the likelihood that this really is a stuck transaction. If that's the
> case then what is the benefit in displaying it as a ms offset from the unix
> epoch, rather than an actual date time?
>
> Thanks,
>
> Tom
>
> On Mon, Aug 31, 2020 at 11:28 PM Guozhang Wang  wrote:
>
> > Thanks Jason, I do not have more comments on the KIP then.
> >
> > On Mon, Aug 31, 2020 at 3:19 PM Jason Gustafson 
> > wrote:
> >
> > > > Hmm, but the "TxnStartOffset" is not included in the
> DescribeProducers
> > > response either?
> > >
> > > Oh, I accidentally called it `CurrentTxnStartTimestamp` in the schema.
> > > Fixed now!
> > >
> > > -Jason
> > >
> > > On Mon, Aug 31, 2020 at 3:04 PM Guozhang Wang 
> > wrote:
> > >
> > > > On Mon, Aug 31, 2020 at 12:28 PM Jason Gustafson  >
> > > > wrote:
> > > >
> > > > > Hey Guozhang,
> > > > >
> > > > > Thanks for the detailed comments. Responses inline:
> > > > >
> > > > > > 1. I'd like to clarify how we can make "--abort" work with old
> > > brokers,
> > > > > since without the additional field "Partitions" the tool needs to
> set
> > > the
> > > > > coordinator epoch correctly instead of "-1"? Arguably that's still
> > > doable
> > > > > but would require different call paths, and it's not clear whether
> > > that's
> > > > > worth doing for old versions.
> > > > >
> > > > > That's a good question. What I had in mind was to write the marker
> > > using
> > > > > the last coordinator epoch that was used by the respective
> > ProducerId.
> > > I
> > > > > realized that I left the coordinator epoch out of the
> > > `DescribeProducers`
> > > > > response, so I have updated the KIP to include it. It is possible
> > that
> > > > > there is no coordinator epoch associated with a given ProducerId
> > (e.g.
> > > if
> > > > > it is the first transaction from that producer), but in this case
> we
> > > can
> > > > > use 0.
> > > > >
> > > > > As for whether this is worth doing, I guess I would be more
> inclined
> > to
> > > > > leave it out if users had a reasonable alternative today to address
> > > this
> > > > > problem.
> > > > >
> > > > > > 2. Why do we have to enforce "DescribeProducers" to be sent to
> only
> > > > > leaders
> > > > > while ListTransactions can be sent to any brokers? Or is it really
> > > > > "ListTransactions to be sent to coordinators only"? From the
> workflow
> > > > > you've described, based on the results back from DescribeProducers,
> > we
> > > > > should just immediately send ListTransactions to the
> > > > > corresponding coordinators based on the collected producer ids,
> > instead
> > > > of
> > > > > trying to send to any brokers right?
> > > > >
> > > > > I'm going to change `DescribeProducers` so that it can be handled
> by
> > > any
> > > > > replica of a topic partition. This was suggested by Lucas in order
> to
> > > > allow
> > > > > this API to be used for replica consistency testing. As far as
> > > > > `ListTransactions`, I was treating this similarly to `ListGroups`.
> > > > Although
> > > > > we know that the coordinators are the leaders of the
> > > __transaction_state
> > > > > partitions, this is more of an implementation detail. From an API
> > > > > perspective, we say that any broker could be a transaction
> > coordinator.
> > > > >
> > > > > > 3. One thing I'm a bit hesitant about is that, is `Describe`
> > > permission
> > > > > on
> > > > > the associated topic sufficient to allow any users to get all
> > producer
> > > > > information writing to the specific topic-partitions including last
> > > > > timestamp, txn-start-timestamp etc, which may be considered
> > sensitive?
> > > > > Should we require "ClusterAction" to only allow operators only?
> > > > >
> > > > > That's a fair point. Do you think `Read` permission would be
> > > reasonable?
> > > > > This is all information that could be obtained by reading the
> topic.
> > > > >
> > > > > Yeah that makes sense.
> > > >
> > > >
> > > > > > 4. From the example it seems "TxnStartOffset" should be included
> in
> > > the
> > > > > DescribeTransaction response schema? Otherwise the user would not
> get
> > > it
> > > > in
> > > > > the following WriteTxnMarker request.
> > > > >
> > > > > The `DescribeTransaction` API is sent to the transaction
> coordinator,
> > > > which
> > > > > does not know the start offset of a transaction in each topic
> > > partition.
> > > > > That is why we need `DescribeProducers`.
> > > > >
> > > >
> > > > Hmm, but 

[jira] [Created] (KAFKA-10472) Consider migrating Topology methods to the Builder pattern

[John Roesler (Jira)]

John Roesler created KAFKA-10472:


 Summary: Consider migrating Topology methods to the Builder pattern
 Key: KAFKA-10472
 URL: https://issues.apache.org/jira/browse/KAFKA-10472
 Project: Kafka
  Issue Type: Improvement
  Components: streams
Reporter: John Roesler


During code review for KIP-478, I got this feedback from [~bbejeck] .

In Topology, we have methods like this:
{code:java}
public synchronized  Topology addGlobalStore(
  final StoreBuilder storeBuilder,
  final String sourceName,
  final TimestampExtractor timestampExtractor,
  final Deserializer keyDeserializer,
  final Deserializer valueDeserializer,
  final String topic,
  final String processorName,
  final ProcessorSupplier stateUpdateSupplier){code}
It would probably be better UX to preset a builder interface like:
{code:java}
public synchronized  Topology addGlobalStore(
  AddGlobalStoreParameters.fromStoreBuilder(storeBuiler)
  .withSourceName(sourceName)
  .withSourceTopic(topic)
  .withTimestampExtractor(timestampExtractor)
  .withKeyDeserializer(keyDeserializer)
  .withValueDeserializer(valueDeserializer)
  .withProcessorName(processorName)
  .withStateUpdateSupplier(stateUpdateSupplier)
){code}
 

Note: new API design proposals should take into account the proposed grammar: 
https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Streams+DSL+Grammar



--
This message was sent by Atlassian Jira
(v8.3.4#803005)

[jira] [Resolved] (KAFKA-10468) Log4jController.getLoggers serialization

[Ismael Juma (Jira)]

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

Ismael Juma resolved KAFKA-10468.
-
Resolution: Duplicate

Duplicate of KAFKA-10403.

> Log4jController.getLoggers serialization
> 
>
> Key: KAFKA-10468
> URL: https://issues.apache.org/jira/browse/KAFKA-10468
> Project: Kafka
>  Issue Type: Bug
>  Components: core
>Reporter: Tom Bentley
>Priority: Minor
>
> {{Log4jController#getLoggers()}} returns a {{java.util.List}} wrapper for a 
> Scala {{List}}, which results in a {{ClassNotFoundException}} on any MBean 
> client which doesn't have the scala wrapper class on its classpath.



--
This message was sent by Atlassian Jira
(v8.3.4#803005)

[jira] [Resolved] (KAFKA-10403) Replace scala collection by java collection in generating MBeans attributes

[Ismael Juma (Jira)]

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

Ismael Juma resolved KAFKA-10403.
-
Fix Version/s: 2.7.0
   Resolution: Fixed

> Replace scala collection by java collection in generating MBeans attributes
> ---
>
> Key: KAFKA-10403
> URL: https://issues.apache.org/jira/browse/KAFKA-10403
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Chia-Ping Tsai
>Assignee: Chia-Ping Tsai
>Priority: Minor
> Fix For: 2.7.0
>
>
> It seems to me the metrics is a "kind" of public interface so users should be 
> able to access metrics of kafka server without scala library.



--
This message was sent by Atlassian Jira
(v8.3.4#803005)

Re: [DISCUSS] Apache Kafka 2.7.0 release

[Dongjin Lee]

Hi Bill,

Could you add the following KIPs to the plan?

- KIP-508: Make Suppression State Queriable

- KIP-653: Upgrade log4j to log4j2


Both KIPs are completely implemented with passing all tests, but not got
reviewed by the committers. Could anyone have a look?

Thanks,
Dongjin

On Wed, Sep 9, 2020 at 8:38 AM Leah Thomas  wrote:

> Hi Bill,
>
> Could you also add KIP-450 to the release plan? It's been merged.
>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-450%3A+Sliding+Window+Aggregations+in+the+DSL
>
> Cheers,
> Leah
>
> On Tue, Sep 8, 2020 at 9:32 AM Bill Bejeck  wrote:
>
> > Hi Bruno,
> >
> > Thanks for letting me know, I've added KIP-662 to the release plan.
> >
> > -Bill
> >
> > On Mon, Sep 7, 2020 at 11:33 AM Bruno Cadonna 
> wrote:
> >
> > > Hi Bill,
> > >
> > > Could you add KIP-662 [1] to the release plan. The KIP has been already
> > > implemented.
> > >
> > > Best,
> > > Bruno
> > >
> > >
> > > [1]
> > >
> > >
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-662%3A+Throw+Exception+when+Source+Topics+of+a+Streams+App+are+Deleted
> > >
> > > On 26.08.20 16:54, Bill Bejeck wrote:
> > > > Greetings All!
> > > >
> > > > I've published a release plan at
> > > >
> > >
> >
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=158872629
> > > .
> > > > I have included all of the KIPs that are currently approved, but I'm
> > > happy
> > > > to make any adjustments as necessary.
> > > >
> > > > The KIP freeze is on September 30 with a target release date of
> > November
> > > 6.
> > > >
> > > > Let me know if there are any objections.
> > > >
> > > > Thanks,
> > > > Bill Bejeck
> > > >
> > > > On Fri, Aug 14, 2020 at 4:01 PM John Roesler 
> > > wrote:
> > > >
> > > >> Thanks, Bill!
> > > >> -John
> > > >>
> > > >> On Thu, 2020-08-13 at 15:19 -0700, Ismael Juma wrote:
> > > >>> Thanks for volunteering Bill. :)
> > > >>>
> > > >>> Ismael
> > > >>>
> > > >>> On Thu, Aug 13, 2020 at 3:13 PM Bill Bejeck 
> > > wrote:
> > > >>>
> > >  Hi All,
> > > 
> > >  I'd like to volunteer to be the release manager for our next
> feature
> > >  release, 2.7. If there are no objections, I'll send out the
> release
> > > >> plan
> > >  soon.
> > > 
> > >  Thanks,
> > >  Bill Bejeck
> > > 
> > > >>
> > > >>
> > > >
> > >
> >
>


-- 
*Dongjin Lee*

*A hitchhiker in the mathematical world.*




*github:  github.com/dongjinleekr
keybase: https://keybase.io/dongjinleekr
linkedin: kr.linkedin.com/in/dongjinleekr
speakerdeck: speakerdeck.com/dongjin
*

Re: [DISCUSS] KIP-664: Provide tooling to detect and abort hanging transactions

[Tom Bentley]

Hi Jason,

The KIP looks good to me, but I had one question. AFAIU the LastTimestamp
column in the output of --describe-producers and --find-hanging is there so
the users of the tool know the txnLastUpdateTimestamp of the
TransactionMetadata and from that and the (max) timeout can infer something
about the likelihood that this really is a stuck transaction. If that's the
case then what is the benefit in displaying it as a ms offset from the unix
epoch, rather than an actual date time?

Thanks,

Tom

On Mon, Aug 31, 2020 at 11:28 PM Guozhang Wang  wrote:

> Thanks Jason, I do not have more comments on the KIP then.
>
> On Mon, Aug 31, 2020 at 3:19 PM Jason Gustafson 
> wrote:
>
> > > Hmm, but the "TxnStartOffset" is not included in the DescribeProducers
> > response either?
> >
> > Oh, I accidentally called it `CurrentTxnStartTimestamp` in the schema.
> > Fixed now!
> >
> > -Jason
> >
> > On Mon, Aug 31, 2020 at 3:04 PM Guozhang Wang 
> wrote:
> >
> > > On Mon, Aug 31, 2020 at 12:28 PM Jason Gustafson 
> > > wrote:
> > >
> > > > Hey Guozhang,
> > > >
> > > > Thanks for the detailed comments. Responses inline:
> > > >
> > > > > 1. I'd like to clarify how we can make "--abort" work with old
> > brokers,
> > > > since without the additional field "Partitions" the tool needs to set
> > the
> > > > coordinator epoch correctly instead of "-1"? Arguably that's still
> > doable
> > > > but would require different call paths, and it's not clear whether
> > that's
> > > > worth doing for old versions.
> > > >
> > > > That's a good question. What I had in mind was to write the marker
> > using
> > > > the last coordinator epoch that was used by the respective
> ProducerId.
> > I
> > > > realized that I left the coordinator epoch out of the
> > `DescribeProducers`
> > > > response, so I have updated the KIP to include it. It is possible
> that
> > > > there is no coordinator epoch associated with a given ProducerId
> (e.g.
> > if
> > > > it is the first transaction from that producer), but in this case we
> > can
> > > > use 0.
> > > >
> > > > As for whether this is worth doing, I guess I would be more inclined
> to
> > > > leave it out if users had a reasonable alternative today to address
> > this
> > > > problem.
> > > >
> > > > > 2. Why do we have to enforce "DescribeProducers" to be sent to only
> > > > leaders
> > > > while ListTransactions can be sent to any brokers? Or is it really
> > > > "ListTransactions to be sent to coordinators only"? From the workflow
> > > > you've described, based on the results back from DescribeProducers,
> we
> > > > should just immediately send ListTransactions to the
> > > > corresponding coordinators based on the collected producer ids,
> instead
> > > of
> > > > trying to send to any brokers right?
> > > >
> > > > I'm going to change `DescribeProducers` so that it can be handled by
> > any
> > > > replica of a topic partition. This was suggested by Lucas in order to
> > > allow
> > > > this API to be used for replica consistency testing. As far as
> > > > `ListTransactions`, I was treating this similarly to `ListGroups`.
> > > Although
> > > > we know that the coordinators are the leaders of the
> > __transaction_state
> > > > partitions, this is more of an implementation detail. From an API
> > > > perspective, we say that any broker could be a transaction
> coordinator.
> > > >
> > > > > 3. One thing I'm a bit hesitant about is that, is `Describe`
> > permission
> > > > on
> > > > the associated topic sufficient to allow any users to get all
> producer
> > > > information writing to the specific topic-partitions including last
> > > > timestamp, txn-start-timestamp etc, which may be considered
> sensitive?
> > > > Should we require "ClusterAction" to only allow operators only?
> > > >
> > > > That's a fair point. Do you think `Read` permission would be
> > reasonable?
> > > > This is all information that could be obtained by reading the topic.
> > > >
> > > > Yeah that makes sense.
> > >
> > >
> > > > > 4. From the example it seems "TxnStartOffset" should be included in
> > the
> > > > DescribeTransaction response schema? Otherwise the user would not get
> > it
> > > in
> > > > the following WriteTxnMarker request.
> > > >
> > > > The `DescribeTransaction` API is sent to the transaction coordinator,
> > > which
> > > > does not know the start offset of a transaction in each topic
> > partition.
> > > > That is why we need `DescribeProducers`.
> > > >
> > >
> > > Hmm, but the "TxnStartOffset" is not included in the DescribeProducers
> > > response either?
> > >
> > >
> > > >
> > > > > 5. It is a bit easier for readers to highlight the added fields in
> > the
> > > > existing WriteTxnMarkerRequest (btw I read is that we are only adding
> > > > "Partitions" with the starting offset, right?). Also as for its
> > response
> > > it
> > > > seems we do not make any schema changes except adding one more
> > potential
> > > > error code "INVALID_TXN_STATE" to it, right? If that's the 

Re: [DISCUSS] KIP-663: API to Start and Shut Down Stream Threads and to Request Closing of Kafka Streams Clients

[Bruno Cadonna]

Hi Matthias and Sophie,

I agree that localThreadsMetadata() can be used here. However, 
localThreadsMetadata() returns all stream threads irrespectively of 
their states. Alive stream threads are specified as being in one of the 
following states: RUNNING, STARTING, PARTITIONS_REVOKED, and 
PARTITIONS_ASSIGNED. Hence, users would need to filter the result of 
localThreadsMetadata(). I thought, it would be neat to have a method 
that hides this filtering and returns the number of alive stream 
threads, because that is the most basic information you might need to 
decide about adding or removing stream threads. For all more advanced 
use cases users should use localThreadsMetadata(). I am also happy with 
removing the method. WDYT?


Best,
Bruno

On 09.09.20 03:51, Matthias J. Sax wrote:

Currently we, don't cleanup dead threads, but the KIP proposes to change
this:


Stream threads that are in state DEAD will be removed from the stream threads 
of a Kafka Streams client.



-Matthias

On 9/8/20 2:37 PM, Sophie Blee-Goldman wrote:

Ah, I forgot about localThreadsMetadata(). In that. case I agree, there's
no reason
to introduce a new method when we can get both the names and number of all
running threads from this.

I assume that we would update localThreadsMetadata to only return currently
alive threads as part of this KIP -- at a quick glance, it seems like we
don't do
any pruning of dead threads at the moment

On Tue, Sep 8, 2020 at 1:58 PM Matthias J. Sax  wrote:


I am not sure if we need a new method? There is already
`localThreadsMetadata()`. What do we gain by adding a new one?

Returning the thread's name (as `Optional`) for both add() and
remove() is fine with me.


-Matthias

On 9/8/20 12:58 PM, Sophie Blee-Goldman wrote:

Sorry Bruno, I think I missed the end of your message with the
numberOfAliveStreamThreads()
proposal. I agree, that would be better than the alternatives I listed.
That said:


They rather suggest that the method returns a list of handles to the

stream threads.

I hadn't thought of that originally, but now that you mention it, this
might be a good idea.
I don't think we should return actual handles on the threads, but maybe a
list of the thread
names rather than a single number of currently alive threads.

Since we seem to think it would be difficult if not impossible to keep
track of the number
of running stream threads, we should apply the same reasoning to the

names

and not
assume the user can/will keep track of every thread returned by
addStreamThread() or
removeStreamThread(). Users should generally take any required action
immediately
after adding/removing the thread -- eg deregistering the thread metrics

--

but it might
still be useful to provide a convenience method listing all of the

current

threads

And of course you could still get the number of threads easily by

invoking

size() on the
returned list (or ordered set?).

On Tue, Sep 8, 2020 at 12:16 PM Bruno Cadonna 

wrote:



Thank you again for the feedback Sophie!

As I tried to point out in my previous e-mail, removing a stream thread
from a Kafka Streams client that does not have alive stream threads is
nothing exceptional for the client per se. However, it can become
exceptional within the context of the user. For example, if users want
to remove a stream thread from a client without alive stream threads
because one if their metrics say so, then this is exceptional in the
context of that user metric not in the context of the Kafka Streams
client. In that case, users should throw an exception and handle it.

Regarding returning null, I do not like to return null because from a
development point of view there is no distinction between returning null
because we have a bug in the code or returning null because there are no
alive stream threads. Additionally, Optional makes it more
explicit that the result could also be empty.

Thank you for the alternative method names! However, with the names you
propose it is not immediately clear that the method returns an amount of
stream threads. They rather suggest that the method returns a list of
handles to the stream threads. I chose to use "aliveStreamThreads" to be
consistent with the client-level metric "alive-stream-threads" which
reports the same number of stream threads that
numberOfAliveStreamThreads() should report. If others also think that
the proposed name in the KIP is too clumsy, I am open to rename it,

though.


Best,
Bruno


On 08.09.20 20:12, Sophie Blee-Goldman wrote:

it's never a good sign when the discussion moves into the vote thread


Hah, sorry, the gmail consolidation of [VOTE] and [DISCUSS] threads

strikes

again.
Thanks for redirecting me Bruno

I suppose it's unfair to expect the callers to keep perfect track of

the

current
   number of stream threads, but it also seems like you shouldn't be

calling

removeStreamThread() when there are no threads left. Either you're just
haphazardly removing threads and could unintentionally slip