[jira] [Resolved] (KAFKA-7930) StreamsResetter makes "changelog" topic naming assumptions

[Matthias J. Sax (JIRA)]

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

Matthias J. Sax resolved KAFKA-7930.

   Resolution: Fixed
Fix Version/s: 2.3.0

> StreamsResetter makes "changelog" topic naming assumptions
> --
>
> Key: KAFKA-7930
> URL: https://issues.apache.org/jira/browse/KAFKA-7930
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams, tools
>Affects Versions: 2.1.0
>Reporter: Murad M
>Priority: Major
>  Labels: features, needs-kip, patch-available, usability
> Fix For: 2.3.0
>
>
> StreamsResetter deletes the topics considered internal. Currently it just 
> checks the naming as per 
> [code|https://github.com/apache/kafka/blob/1aae604861068bb7337d4972c9dcc0c0a99c374d/core/src/main/scala/kafka/tools/StreamsResetter.java#L660].
>  If assumption is wrong (either topic prefix or suffix), tool becomes useless 
> if aware even dangerous if not. Probably better either:
>  * naming assumption should be optional and supply internal topics with 
> argument (--internal-topics)
>  * deletion could be optional (--no-delete-internal)
>  * ignore topics which are included in list of --input-topics
> Faced this, when was trying to reset applications with GlobalKTable topics 
> named as *-changelog. Such topics sometimes are not desirable for deletion.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Build failed in Jenkins: kafka-trunk-jdk8 #3402

[Apache Jenkins Server]

See 


Changes:

[wangguoz] KAFKA-4730: Streams does not have an in-memory windowed store (#6239)

--
[...truncated 4.28 MB...]

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

[jira] [Resolved] (KAFKA-7283) mmap indexes lazily and skip sanity check for segments below recovery point

[Jun Rao (JIRA)]

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

Jun Rao resolved KAFKA-7283.

   Resolution: Fixed
Fix Version/s: 2.3.0

Merged the PR to trunk.

> mmap indexes lazily and skip sanity check for segments below recovery point
> ---
>
> Key: KAFKA-7283
> URL: https://issues.apache.org/jira/browse/KAFKA-7283
> Project: Kafka
>  Issue Type: New Feature
>Reporter: Zhanxiang (Patrick) Huang
>Assignee: Zhanxiang (Patrick) Huang
>Priority: Major
> Fix For: 2.3.0
>
>
> This is a follow-up ticket for KIP-263.
> Currently broker will mmap the index files, read the length as well as the 
> last entry of the file, and sanity check index files of all log segments in 
> the log directory after the broker is started. These operations can be slow 
> because broker needs to open index file and read data into page cache. In 
> this case, the time to restart a broker will increase proportional to the 
> number of segments in the log directory.
> Per the KIP discussion, we think we can skip sanity check for segments below 
> the recovery point since Kafka does not provide guarantee for segments 
> already flushed to disk and sanity checking only index file benefits little 
> when the segment is also corrupted because of disk failure. Therefore, we can 
> make the following changes to improve broker startup time:
>  # Mmap the index file and populate fields of the index file on-demand rather 
> than performing costly disk operations when creating the index object on 
> broker startup.
>  # Skip sanity checks on indexes of segments below the recovery point.
> With these changes, the broker startup time will increase only proportional 
> to the number of partitions in the log directly after cleaned shutdown 
> because only active segments are mmaped and sanity checked.
>  
>  



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (KAFKA-7967) Kafka Streams: some values in statestore rollback to old value

[Ziming Dong (JIRA)]

Ziming Dong created KAFKA-7967:
--

 Summary: Kafka Streams: some values in statestore rollback to old 
value
 Key: KAFKA-7967
 URL: https://issues.apache.org/jira/browse/KAFKA-7967
 Project: Kafka
  Issue Type: Bug
Affects Versions: 2.1.0
Reporter: Ziming Dong


We are using kafka streams 2.1.0, we use both persistentKeyValueStore 
statestore and persistentWindowStore statestore. We found sometimes both types 
of statestore could `fetch` old values instead of newly updated values. We 
didn't find any logs except INFO level logs, also there is no rebalance log 
which indicates it's not a rebalance bug. The bug happened no more than one 
time each week, but many records were affected each time, and we didn't find a 
way to reproduce it manually.

For example, the issue may happen like this:
 # got value 1 from key 1
 # update value 2 to key 1
 # got value 2 from key 1
 # update value 3 to key 1
 # got value 1 from key 1(something wrong!!)
 # update value 2 to key 1

there is only one type log as follow

 
{code:java}
2019-02-19x14:20:00x xx INFO [org.apache.kafka.clients.FetchSessionHandler] 
[xxx-streams-xx-xxx--xxx-xx-StreamThread-1] [Consumer 
clientId=x--xxx-xxx--x-StreamThread-1-consumer, 
groupId=x] Node 2 was unable to process the fetch request with 
(sessionId=1998942517, epoch=4357): INVALID_FETCH_SESSION_EPOCH.
{code}
 



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Build failed in Jenkins: kafka-trunk-jdk11 #301

[Apache Jenkins Server]

See 


Changes:

[github] HOT_FIX: Change flag so plain RocksDB instance returned (#6297)

[mjsax] MINOR: refactor topic check to make sure all topics exist by name vs

--
[...truncated 2.33 MB...]
at 
org.gradle.internal.event.BroadcastDispatch$CompositeDispatch.dispatch(BroadcastDispatch.java:234)
at 
org.gradle.internal.event.ListenerBroadcast.dispatch(ListenerBroadcast.java:140)
at 
org.gradle.internal.event.ListenerBroadcast.dispatch(ListenerBroadcast.java:37)
at 
org.gradle.internal.dispatch.ProxyDispatchAdapter$DispatchingInvocationHandler.invoke(ProxyDispatchAdapter.java:93)
at com.sun.proxy.$Proxy85.output(Unknown Source)
at 
org.gradle.api.internal.tasks.testing.results.StateTrackingTestResultProcessor.output(StateTrackingTestResultProcessor.java:112)
at 
org.gradle.api.internal.tasks.testing.results.AttachParentTestResultProcessor.output(AttachParentTestResultProcessor.java:48)
at jdk.internal.reflect.GeneratedMethodAccessor349.invoke(Unknown 
Source)
at 
java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:566)
at 
org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:35)
at 
org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24)
at 
org.gradle.internal.dispatch.FailureHandlingDispatch.dispatch(FailureHandlingDispatch.java:29)
at 
org.gradle.internal.dispatch.AsyncDispatch.dispatchMessages(AsyncDispatch.java:87)
at 
org.gradle.internal.dispatch.AsyncDispatch.access$000(AsyncDispatch.java:36)
at 
org.gradle.internal.dispatch.AsyncDispatch$1.run(AsyncDispatch.java:71)
at 
org.gradle.internal.concurrent.InterruptibleRunnable.run(InterruptibleRunnable.java:42)
at 
org.gradle.internal.operations.CurrentBuildOperationPreservingRunnable.run(CurrentBuildOperationPreservingRunnable.java:42)
at 
org.gradle.internal.concurrent.ExecutorPolicy$CatchAndRecordFailures.onExecute(ExecutorPolicy.java:63)
at 
org.gradle.internal.concurrent.ManagedExecutorImpl$1.run(ManagedExecutorImpl.java:46)
at 
java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at 
java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at 
org.gradle.internal.concurrent.ThreadFactoryImpl$ManagedThreadRunnable.run(ThreadFactoryImpl.java:55)
at java.base/java.lang.Thread.run(Thread.java:834)
ERROR: Failed to write output for test null.Gradle Test Executor 20
java.lang.NullPointerException: Cannot invoke method write() on null object
at 
org.codehaus.groovy.runtime.NullObject.invokeMethod(NullObject.java:91)
at 
org.codehaus.groovy.runtime.callsite.PogoMetaClassSite.call(PogoMetaClassSite.java:47)
at 
org.codehaus.groovy.runtime.callsite.CallSiteArray.defaultCall(CallSiteArray.java:47)
at 
org.codehaus.groovy.runtime.callsite.NullCallSite.call(NullCallSite.java:34)
at 
org.codehaus.groovy.runtime.callsite.CallSiteArray.defaultCall(CallSiteArray.java:47)
at 
org.codehaus.groovy.runtime.callsite.PogoMetaClassSite.call(PogoMetaClassSite.java:56)
at 
org.codehaus.groovy.runtime.callsite.AbstractCallSite.call(AbstractCallSite.java:127)
at 
build_av7tdo8a1kfk9insmdebvqzp9$_run_closure5$_closure74$_closure107.doCall(:244)
at jdk.internal.reflect.GeneratedMethodAccessor350.invoke(Unknown 
Source)
at 
java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:566)
at 
org.codehaus.groovy.reflection.CachedMethod.invoke(CachedMethod.java:104)
at groovy.lang.MetaMethod.doMethodInvoke(MetaMethod.java:326)
at 
org.codehaus.groovy.runtime.metaclass.ClosureMetaClass.invokeMethod(ClosureMetaClass.java:264)
at groovy.lang.MetaClassImpl.invokeMethod(MetaClassImpl.java:1041)
at groovy.lang.Closure.call(Closure.java:411)
at 
org.gradle.listener.ClosureBackedMethodInvocationDispatch.dispatch(ClosureBackedMethodInvocationDispatch.java:40)
at 
org.gradle.listener.ClosureBackedMethodInvocationDispatch.dispatch(ClosureBackedMethodInvocationDispatch.java:25)
at 
org.gradle.internal.event.AbstractBroadcastDispatch.dispatch(AbstractBroadcastDispatch.java:42)
at 
org.gradle.internal.event.BroadcastDispatch$SingletonDispatch.dispatch(BroadcastDispatch.java:230)
at 
org.gradle.internal.event.BroadcastDispatch$SingletonDispatch.dispatch(BroadcastDispatch.java:149)
at 

Jenkins build is back to normal : kafka-2.2-jdk8 #26

[Apache Jenkins Server]

See 

Re: [DISCUSS] KIP-300: Add Windowed KTable API in StreamsBuilder

[Boyang Chen]

Great, thank you Matthias!



From: Matthias J. Sax 
Sent: Thursday, February 21, 2019 10:55 AM
To: dev@kafka.apache.org
Subject: Re: [DISCUSS] KIP-300: Add Windowed KTable API in StreamsBuilder

Thanks. LGTM.

-Matthias

On 1/20/19 8:39 PM, Boyang Chen wrote:
> Hey Mattihas,
>
> I have addressed the comments in KIP. Feel free to take another look.
>
> Also you are right, those are implementation details that we could discuss in 
> diff 
>
> Boyang
>
> 
> From: Matthias J. Sax 
> Sent: Saturday, January 19, 2019 3:16 AM
> To: dev@kafka.apache.org
> Subject: Re: [DISCUSS] KIP-300: Add Windowed KTable API in StreamsBuilder
>
> Thank Boyang!
>
>>> I think it should be good to just extend ConsumedInternal and 
>>> MaterializedInternal with window size, and keep
>>> external API clean. Just so you know it would be even more messy for 
>>> internal implementation if we don't carry
>>> the window size around within existing data struct.
>
> I cannot follow here. But because this is internal stuff anyway, I would
> prefer to discuss this on the PR instead of the mailing list.
>
>
> -Matthias
>
> On 1/18/19 10:58 AM, Boyang Chen wrote:
>> Hey Matthias,
>>
>> thanks a lot for the comments!
>>
>> It seems that `windowSize` is a mandatory argument for windowed-tables,
>> thus all overload should have the first two parameters being `String
>> topic` and `Duration windowSize`.
>> Yep, that sounds good to me.
>>
>> For session-tables, there should be no `windowSize` parameter because
>> each session can have a different size and as a matter of fact, both the
>> window start and window end timestamp are contained in the key anyway
>> for this reason. (This is different to time windows as the KIP mentions.)
>> Good suggestion, I think we should be able to skip the windowsize for 
>> session store.
>>
>> Thus, I don't think that there is any need to extend `Consumed` or
>> `Materialized` -- in contrast, extending both as suggested would result
>> in bad API, because those new methods would be available for
>> key-value-tables, too.
>> I think it should be good to just extend ConsumedInternal and 
>> MaterializedInternal with window size, and keep
>> external API clean. Just so you know it would be even more messy for 
>> internal implementation if we don't carry
>> the window size around within existing data struct.
>>
>> About generic types: why is `windowedTable()` using `Consumers`
>> while `sessionTable` is using `Consumed>`? The KIP
>> mentions that we can wrap provided key-serdes automatically with
>> corresponding window serdes. Thus, it seems the correct type should be `K`?
>> Yes that's a typo, and I already fixed it.
>>
>> I will let you know when the KIP updates are done.
>>
>> Best,
>> Boyang
>> 
>> From: Matthias J. Sax 
>> Sent: Thursday, January 17, 2019 7:52 AM
>> To: dev@kafka.apache.org
>> Subject: Re: [DISCUSS] KIP-300: Add Windowed KTable API in StreamsBuilder
>>
>> Couple of follow up comment on the KIP:
>>
>> It seems that `windowSize` is a mandatory argument for windowed-tables,
>> thus all overload should have the first two parameters being `String
>> topic` and `Duration windowSize`.
>>
>> For session-tables, there should be no `windowSize` parameter because
>> each session can have a different size and as a matter of fact, both the
>> window start and window end timestamp are contained in the key anyway
>> for this reason. (This is different to time windows as the KIP mentions.)
>>
>> Thus, I don't think that there is any need to extend `Consumed` or
>> `Materialized` -- in contrast, extending both as suggested would result
>> in bad API, because those new methods would be available for
>> key-value-tables, too.
>>
>> About generic types: why is `windowedTable()` using `Consumers`
>> while `sessionTable` is using `Consumed>`? The KIP
>> mentions that we can wrap provided key-serdes automatically with
>> corresponding window serdes. Thus, it seems the correct type should be `K`?
>>
>>
>> -Matthias
>>
>>
>> On 1/12/19 8:35 PM, Boyang Chen wrote:
>>> Hey Matthias,
>>>
>>> thanks for taking a look! It would be great to see this pushed in 2.2. 
>>> Depending on the tight timeline, I hope to at least get the KIP approved so 
>>> that we don't see back and forth again as the KTable API has been 
>>> constantly changing. I couldn't guarantee the implementation timeline until 
>>> we agree on the updated high level APIs first. Does that make sense?
>>>
>>> Best,
>>> Boyang
>>> 
>>> From: Matthias J. Sax 
>>> Sent: Sunday, January 13, 2019 10:53 AM
>>> To: dev@kafka.apache.org
>>> Subject: Re: [DISCUSS] KIP-300: Add Windowed KTable API in StreamsBuilder
>>>
>>> Do you want to get this into 2.2 release? KIP deadline is 1/24, so quite
>>> soon.
>>>
>>> Overall, the KIP is very useful. I can review again in more details if
>>> you aim for 2.2 -- did you address all previous comment 

Re: [VOTE] KIP-412: Extend Admin API to support dynamic application log levels

[Harsha]

+1 (binding).

Thanks,
Harsha

On Tue, Feb 19, 2019, at 7:53 AM, Andrew Schofield wrote:
> Thanks for the KIP.
> 
> +1 (non-binding)
> 
> On 18/02/2019, 12:48, "Stanislav Kozlovski"  wrote:
> 
> Hey everybody, I'm starting a VOTE thread for KIP-412. This feature should
> significantly improve the flexibility and ease in debugging Kafka in run
> time
> 
> KIP-412 -
> 
> https://nam03.safelinks.protection.outlook.com/?url=https%3A%2F%2Fcwiki.apache.org%2Fconfluence%2Fdisplay%2FKAFKA%2FKIP-412%253A%2BExtend%2BAdmin%2BAPI%2Bto%2Bsupport%2Bdynamic%2Bapplication%2Blog%2Blevelsdata=02%7C01%7C%7C69bc63a9d7864e25ec3c08d69596eec4%7C84df9e7fe9f640afb435%7C1%7C0%7C636860872825557120sdata=XAnMhy6EPC7JkB77NBBhLR%2FvE7XrTutuS5Rlt%2FDpwfU%3Dreserved=0
> 
> 
> -- 
> Best,
> Stanislav
> 
> 
>

Build failed in Jenkins: kafka-trunk-jdk8 #3401

[Apache Jenkins Server]

See 


Changes:

[rajinisivaram] KAFKA-7945: Calc refresh time correctly when token created in 
the past

[github] HOT_FIX: Change flag so plain RocksDB instance returned (#6297)

[mjsax] MINOR: refactor topic check to make sure all topics exist by name vs

--
[...truncated 2.31 MB...]
> Task :streams:upgrade-system-tests-0102:compileTestJava
> Task :streams:upgrade-system-tests-0102:processTestResources NO-SOURCE

> Task :streams:streams-scala:test

org.apache.kafka.streams.scala.StreamToTableJoinScalaIntegrationTestImplicitSerdes
 > testShouldCountClicksPerRegionWithNamedRepartitionTopic STARTED

org.apache.kafka.streams.scala.StreamToTableJoinScalaIntegrationTestImplicitSerdes
 > testShouldCountClicksPerRegionWithNamedRepartitionTopic PASSED

org.apache.kafka.streams.scala.StreamToTableJoinScalaIntegrationTestImplicitSerdes
 > testShouldCountClicksPerRegionJava STARTED

org.apache.kafka.streams.scala.StreamToTableJoinScalaIntegrationTestImplicitSerdes
 > testShouldCountClicksPerRegionJava PASSED

org.apache.kafka.streams.scala.StreamToTableJoinScalaIntegrationTestImplicitSerdes
 > testShouldCountClicksPerRegion STARTED

org.apache.kafka.streams.scala.StreamToTableJoinScalaIntegrationTestImplicitSerdes
 > testShouldCountClicksPerRegion PASSED

org.apache.kafka.streams.scala.TopologyTest > 
shouldBuildIdenticalTopologyInJavaNScalaJoin STARTED

org.apache.kafka.streams.scala.TopologyTest > 
shouldBuildIdenticalTopologyInJavaNScalaJoin PASSED

org.apache.kafka.streams.scala.TopologyTest > 
shouldBuildIdenticalTopologyInJavaNScalaSimple STARTED

org.apache.kafka.streams.scala.TopologyTest > 
shouldBuildIdenticalTopologyInJavaNScalaSimple PASSED

org.apache.kafka.streams.scala.TopologyTest > 
shouldBuildIdenticalTopologyInJavaNScalaAggregate STARTED

org.apache.kafka.streams.scala.TopologyTest > 
shouldBuildIdenticalTopologyInJavaNScalaAggregate PASSED

org.apache.kafka.streams.scala.TopologyTest > 
shouldBuildIdenticalTopologyInJavaNScalaTransform STARTED

org.apache.kafka.streams.scala.TopologyTest > 
shouldBuildIdenticalTopologyInJavaNScalaTransform PASSED

org.apache.kafka.streams.scala.WordCountTest > testShouldCountWordsMaterialized 
STARTED

org.apache.kafka.streams.scala.WordCountTest > testShouldCountWordsMaterialized 
PASSED

org.apache.kafka.streams.scala.WordCountTest > testShouldCountWordsJava STARTED

org.apache.kafka.streams.scala.WordCountTest > testShouldCountWordsJava PASSED

org.apache.kafka.streams.scala.WordCountTest > testShouldCountWords STARTED

org.apache.kafka.streams.scala.WordCountTest > testShouldCountWords PASSED

org.apache.kafka.streams.scala.kstream.MaterializedTest > Create a Materialized 
should create a Materialized with Serdes STARTED

org.apache.kafka.streams.scala.kstream.MaterializedTest > Create a Materialized 
should create a Materialized with Serdes PASSED

org.apache.kafka.streams.scala.kstream.MaterializedTest > Create a Materialize 
with a store name should create a Materialized with Serdes and a store name 
STARTED

org.apache.kafka.streams.scala.kstream.MaterializedTest > Create a Materialize 
with a store name should create a Materialized with Serdes and a store name 
PASSED

org.apache.kafka.streams.scala.kstream.MaterializedTest > Create a Materialize 
with a window store supplier should create a Materialized with Serdes and a 
store supplier STARTED

org.apache.kafka.streams.scala.kstream.MaterializedTest > Create a Materialize 
with a window store supplier should create a Materialized with Serdes and a 
store supplier PASSED

org.apache.kafka.streams.scala.kstream.MaterializedTest > Create a Materialize 
with a key value store supplier should create a Materialized with Serdes and a 
store supplier STARTED

org.apache.kafka.streams.scala.kstream.MaterializedTest > Create a Materialize 
with a key value store supplier should create a Materialized with Serdes and a 
store supplier PASSED

org.apache.kafka.streams.scala.kstream.MaterializedTest > Create a Materialize 
with a session store supplier should create a Materialized with Serdes and a 
store supplier STARTED

org.apache.kafka.streams.scala.kstream.MaterializedTest > Create a Materialize 
with a session store supplier should create a Materialized with Serdes and a 
store supplier PASSED

org.apache.kafka.streams.scala.kstream.KTableTest > filter a KTable should 
filter records satisfying the predicate STARTED

org.apache.kafka.streams.scala.kstream.KTableTest > filter a KTable should 
filter records satisfying the predicate PASSED

org.apache.kafka.streams.scala.kstream.KTableTest > filterNot a KTable should 
filter records not satisfying the predicate STARTED

org.apache.kafka.streams.scala.kstream.KTableTest > filterNot a KTable should 
filter records not satisfying the predicate PASSED

org.apache.kafka.streams.scala.kstream.KTableTest > join 2 KTables should join 
correctly 

[jira] [Resolved] (KAFKA-4730) Streams does not have an in-memory windowed store

[Guozhang Wang (JIRA)]

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

Guozhang Wang resolved KAFKA-4730.
--
   Resolution: Fixed
Fix Version/s: 2.3.0

> Streams does not have an in-memory windowed store
> -
>
> Key: KAFKA-4730
> URL: https://issues.apache.org/jira/browse/KAFKA-4730
> Project: Kafka
>  Issue Type: New Feature
>  Components: streams
>Affects Versions: 0.10.2.0
>Reporter: Eno Thereska
>Assignee: Sophie Blee-Goldman
>Priority: Major
> Fix For: 2.3.0
>
>
> Streams has windowed persistent stores (e.g., see PersistentKeyValueFactory 
> interface with "windowed" method), however it does not allow for windowed 
> in-memory stores (e.g., see InMemoryKeyValueFactory interface). 
> In addition to the interface not allowing it, streams does not actually have 
> an implementation of an in-memory windowed store.
> The implications are that operations that require windowing cannot use 
> in-memory stores. 



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Re: [VOTE] KIP-300: Add Windowed KTable API in StreamsBuilder

[Matthias J. Sax]

Thanks for the KIP Boyang!


+1 (binding)


-Matthias

On 1/16/19 9:15 AM, Boyang Chen wrote:
> Hey friends,
> 
> I would like to start the vote thread for KIP-300 so that we could agree on 
> the high level API. Feel free to continue making comment on the discussion 
> thread.
> 
> Best,
> Boyang
> 



signature.asc
Description: OpenPGP digital signature

Re: [DISCUSS] KIP-300: Add Windowed KTable API in StreamsBuilder

[Matthias J. Sax]

Thanks. LGTM.

-Matthias

On 1/20/19 8:39 PM, Boyang Chen wrote:
> Hey Mattihas,
> 
> I have addressed the comments in KIP. Feel free to take another look.
> 
> Also you are right, those are implementation details that we could discuss in 
> diff 
> 
> Boyang
> 
> 
> From: Matthias J. Sax 
> Sent: Saturday, January 19, 2019 3:16 AM
> To: dev@kafka.apache.org
> Subject: Re: [DISCUSS] KIP-300: Add Windowed KTable API in StreamsBuilder
> 
> Thank Boyang!
> 
>>> I think it should be good to just extend ConsumedInternal and 
>>> MaterializedInternal with window size, and keep
>>> external API clean. Just so you know it would be even more messy for 
>>> internal implementation if we don't carry
>>> the window size around within existing data struct.
> 
> I cannot follow here. But because this is internal stuff anyway, I would
> prefer to discuss this on the PR instead of the mailing list.
> 
> 
> -Matthias
> 
> On 1/18/19 10:58 AM, Boyang Chen wrote:
>> Hey Matthias,
>>
>> thanks a lot for the comments!
>>
>> It seems that `windowSize` is a mandatory argument for windowed-tables,
>> thus all overload should have the first two parameters being `String
>> topic` and `Duration windowSize`.
>> Yep, that sounds good to me.
>>
>> For session-tables, there should be no `windowSize` parameter because
>> each session can have a different size and as a matter of fact, both the
>> window start and window end timestamp are contained in the key anyway
>> for this reason. (This is different to time windows as the KIP mentions.)
>> Good suggestion, I think we should be able to skip the windowsize for 
>> session store.
>>
>> Thus, I don't think that there is any need to extend `Consumed` or
>> `Materialized` -- in contrast, extending both as suggested would result
>> in bad API, because those new methods would be available for
>> key-value-tables, too.
>> I think it should be good to just extend ConsumedInternal and 
>> MaterializedInternal with window size, and keep
>> external API clean. Just so you know it would be even more messy for 
>> internal implementation if we don't carry
>> the window size around within existing data struct.
>>
>> About generic types: why is `windowedTable()` using `Consumers`
>> while `sessionTable` is using `Consumed>`? The KIP
>> mentions that we can wrap provided key-serdes automatically with
>> corresponding window serdes. Thus, it seems the correct type should be `K`?
>> Yes that's a typo, and I already fixed it.
>>
>> I will let you know when the KIP updates are done.
>>
>> Best,
>> Boyang
>> 
>> From: Matthias J. Sax 
>> Sent: Thursday, January 17, 2019 7:52 AM
>> To: dev@kafka.apache.org
>> Subject: Re: [DISCUSS] KIP-300: Add Windowed KTable API in StreamsBuilder
>>
>> Couple of follow up comment on the KIP:
>>
>> It seems that `windowSize` is a mandatory argument for windowed-tables,
>> thus all overload should have the first two parameters being `String
>> topic` and `Duration windowSize`.
>>
>> For session-tables, there should be no `windowSize` parameter because
>> each session can have a different size and as a matter of fact, both the
>> window start and window end timestamp are contained in the key anyway
>> for this reason. (This is different to time windows as the KIP mentions.)
>>
>> Thus, I don't think that there is any need to extend `Consumed` or
>> `Materialized` -- in contrast, extending both as suggested would result
>> in bad API, because those new methods would be available for
>> key-value-tables, too.
>>
>> About generic types: why is `windowedTable()` using `Consumers`
>> while `sessionTable` is using `Consumed>`? The KIP
>> mentions that we can wrap provided key-serdes automatically with
>> corresponding window serdes. Thus, it seems the correct type should be `K`?
>>
>>
>> -Matthias
>>
>>
>> On 1/12/19 8:35 PM, Boyang Chen wrote:
>>> Hey Matthias,
>>>
>>> thanks for taking a look! It would be great to see this pushed in 2.2. 
>>> Depending on the tight timeline, I hope to at least get the KIP approved so 
>>> that we don't see back and forth again as the KTable API has been 
>>> constantly changing. I couldn't guarantee the implementation timeline until 
>>> we agree on the updated high level APIs first. Does that make sense?
>>>
>>> Best,
>>> Boyang
>>> 
>>> From: Matthias J. Sax 
>>> Sent: Sunday, January 13, 2019 10:53 AM
>>> To: dev@kafka.apache.org
>>> Subject: Re: [DISCUSS] KIP-300: Add Windowed KTable API in StreamsBuilder
>>>
>>> Do you want to get this into 2.2 release? KIP deadline is 1/24, so quite
>>> soon.
>>>
>>> Overall, the KIP is very useful. I can review again in more details if
>>> you aim for 2.2 -- did you address all previous comment about the KIP
>>> already?
>>>
>>>
>>> -Matthias
>>>
>>>
>>>
>>> On 1/8/19 2:50 PM, Boyang Chen wrote:
 Hey folks,

 I would like to start a discussion thread on adding new time/session 
 windowed KTable 

Re: [DISCUSS] KIP-307: Allow to define custom processor names with KStreams DSL

[Matthias J. Sax]

Florian,

thanks for updating the KIP (and no worries for late reply -- 2.2
release kept us busy anyway...). Overall LGTM.

Just some nits:


KStream-Table:

Do we need to list the existing stream-globalTable join methods in the
first table (thought it should only contain new/changing methods).

typo: `join(GlobalKTbale, KeyValueMapper, ValueJoiner, Named)`

`leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner)` is missing the new
`Named` parameter.

`static Joined#named(final String name)`
 -> should be `#as(...)` instead of `named(...)`

flatTransform() is missing (cf. KIP-313)



KTable-table:

`Suppressed#withName(String)`
 -> should we change this to `#as(...)` too (similar to `named()`)



-Matthias



On 1/25/19 9:49 AM, Matthias J. Sax wrote:
> I was reading the KIP again, and there are still some open question and
> inconsistencies:
> 
> For example for `KGroupedStream#count(Named)` the KIP says, that only
> the processor will be named, while the state store name will be `PREFIX
> + COUNT` (ie, an auto-generated name). Additionally, for
> `KGroupedStream#count(Named, Materialized)` the processor will be named
> according to `Named` and the store will be named according to
> `Materialized.as()`. So far so good. It implies that naming the
> processor and naming the store are independent. (This pattern is applied
> to all aggregation functions, for KStream and KTable).
> 
> However, for `KTable#filter(Predicate, Named)` the KIP says, the
> processor name and the store name are set. This sound wrong (ie,
> inconsistent with the first paragraph from above), because there is also
> `KTable#filter(Predicate, Named, Materialized)`. Also note, for the
> first operator, the store might not be materialized to at all. (This
> issue is there for all KTable operators -- stateless and stateful).
> 
> Finally, there is the following statement in the KIP:
> 
>> Also, note that for all methods accepting a Materialized argument, if no 
>> state store named is provided then the node named will be used to generate a 
>> one. The state store name will be the node name suffixed with "-table".
> 
> 
> This contradict the non-naming of stores from the very beginning.
> 
> 
> Also, the KIP still contains the question about `join(GlobalKTable,
> KeyValueMapper, ValueJoiner)` and `leftJoin(GlobalKTable,
> KeyValueMapper, ValueJoiner)`. I think a consistent approach would be to
> add one overload each that takes a `Named` parameter.
> 
> 
> Thoughts?
> 
> 
> -Matthias
> 
> 
> On 1/17/19 2:56 PM, Bill Bejeck wrote:
>> +1 for me on Guozhang's proposal for changes to Joined.
>>
>> Thanks,
>> Bill
>>
>> On Thu, Jan 17, 2019 at 5:55 PM Matthias J. Sax 
>> wrote:
>>
>>> Thanks for all the follow up comments!
>>>
>>> As I mentioned earlier, I am ok with adding overloads instead of using
>>> Materialized to specify the processor name. Seems this is what the
>>> majority of people prefers.
>>>
>>> I am also +1 on Guozhang's suggestion to deprecate `static
>>> Joined#named()` and replace it with `static Joined#as` for consistency
>>> and to deprecate getter `Joined#name()` for removal and introduce
>>> `JoinedInternal` to access the name.
>>>
>>> @Guozhang: the vote is already up :)
>>>
>>>
>>> -Matthias
>>>
>>> On 1/17/19 2:45 PM, Guozhang Wang wrote:
 Wow that's a lot of discussions in 6 days! :) Just catching up and
>>> sharing
 my two cents here:

 1. Materialized: I'm inclined to not let Materialized extending Named and
 add the overload as well. All the rationales have been very well
>>> summarized
 before. Just to emphasize on John's points: Materialized is considered as
 the control object being leveraged by the optimization framework to
 determine if the state store should be physically materialized or not. So
 let's say if the user does not want to query the store (hence it can just
 be locally materialized), but still want to name the processor, they need
 to do either "count(Materialized.as(null).withName("processorName"));" or
 "count(Named.as("processorName"));" and neither of it is a bit hard to
 educate to users, and hence it looks that an overload function with two
 parameters are easier to understand.

 2. As for `NamedOperation`: I've left a comment about it before, i.e. "1)
 Regarding the interface / function name, I'd propose we call the
>>> interface
 `NamedOperation` which would be implemented by Produced / Consumed /
 Printed / Joined / Grouped / Suppressed (note I intentionally exclude
 Materialized here since its semantics is quite), and have the default
>>> class
 that implements `NamedOperation` as `Named`, which would be used in our
 adding overload functions. The main reason is to have consistency in
 naming." And I think I'm on the same page with John with his more
>>> detailed
 proposal.

 3. As for `Joined`: I actually would suggest we bite the bullet and
>>> remove
 it as well, because we are 

Re: [DISCUSS] KIP-424: Allow suppression of intermediate events based on wall clock time

[Matthias J. Sax]

Jonathan,

thanks for the KIP. Corner case question:

What happens if an application is stopped an restarted?

 - Should suppress() flush all records (would be _before_ the time elapsed)?
 - Or should it preserve buffered records and reload on restart? For
this case, should the record be flushed on reload (elapsed time is
unknown) or should we reset the timer to zero?


What is unclear to me atm, is the use-case you anticipate. If you assume
a live run of an applications, event-time and processing-time should be
fairly identical (at least with regard to data rates). Thus, suppress()
on event-time should give you about the same behavior as wall-clock
time? If you disagree, can you elaborate?

This leave the case for data reprocessing, for which event-time advances
much faster than wall-clock time. Is this the target use-case?


About the implementation: checking wall-clock time is an expensive
system call, so I am little worried about run-time overhead. This seems
not to be an implementation detail and thus, it might be worth to
includes is in the discussion. The question is, how strict the guarantee
when records should be flushed should be. Assume you set a timer of 1
seconds, and you have a data rate of 1000 records per second, with each
record arriving one ms after the other all each with different key. To
flush this data "correctly" we would need to check wall-clock time very
millisecond... Thoughts?

(We don't need to dive into all details, but a high level discussion
about the desired algorithm and guarantees would be good to have IMHO.)





-Matthias


On 1/30/19 12:16 PM, John Roesler wrote:
> Hi Jonathan,
> 
> Thanks for the KIP!
> 
> I think all the reviewers are heads-down right now reviewing code for the
> imminent 2.2 release, so this discussion may not get much traffic over the
> next couple of weeks. You might want to just keep bumping it once a week or
> so until people start finding time to review and respond.
> 
> Also, This message got marked as spam for me (which happens for mailing
> list messages sometimes, for some reason). I'm hoping that this response
> will hoist it into peoples' inboxes...
> 
> Thanks again for your work on this issue, and I look forward to the
> discussion!
> -John
> 
> On Wed, Jan 30, 2019 at 12:24 AM jonathangor...@newrelic.com <
> jonathangor...@newrelic.com> wrote:
> 
>> Hi all,
>>
>> I just published KIP-424: Allow suppression of intermediate events based
>> on wall clock time
>>
>>
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-424%3A+Allow+suppression+of+intermediate+events+based+on+wall+clock+time
>>
>> I am eager to hear your feedback and concerns. Thanks John Roesler for
>> your guidance in shaping my first KIP!
>>
>> I look forward to working with the Kafka community to see this through,
>>
>> Jonathan
>>
>>
>>
> 



signature.asc
Description: OpenPGP digital signature

Re: Statestore restoration & scaling questions - possible KIP as well.

[Guozhang Wang]

Hello Adam,

Sorry for being late replying on this thread, I've put my comments inlined
below.

On Sun, Feb 3, 2019 at 7:34 AM Adam Bellemare 
wrote:

> Hey Folks
>
> I have a few questions around the operations of stateful processing while
> scaling nodes up/down, and a possible KIP in question #4. Most of them have
> to do with task processing during rebuilding of state stores after scaling
> nodes up.
>
> Scenario:
> Single node/thread, processing 2 topics (10 partitions each):
> User event topic (events) - ie: key:userId, value: ProductId
> Product topic (entity) - ie: key: ProductId, value: productData
>
> My topology looks like this:
>
> KTable productTable = ... //materialize from product topic
>
> KStream output = userStream
> .map(x => (x.value, x.key) ) //Swap the key and value around
> .join(productTable, ... ) //Joiner is not relevant here
> .to(...)  //Send it to some output topic
>
>
> Here are my questions:
> 1) If I scale the processing node count up, partitions will be rebalanced
> to the new node. Does processing continue as normal on the original node,
> while the new node's processing is paused as the internal state stores are
> rebuilt/reloaded? From my reading of the code (and own experience) I
> believe this to be the case, but I am just curious in case I missed
> something.
>
>
With 2 topics and 10 partitions each, assuming the default PartitionGrouper
is used, there should be a total of 20 tasks (10 tasks for map which will
send to an internal repartition topic, and 10 tasks for doing the join)
created since these two topics are co-partitioned for joins.

For example, task-0 would be processing the join from
user-topic-partition-0 and product-topic-partition-0, and so on.

With a single thread, all of these 20 tasks will be allocated to this
thread, which would process them in an iterative manner. Note that since
each task has its own state store (e.g. product-state-store-0 for task-0,
etc), it means this thread will host all the 10 sets of state stores as
well (note for the 10 mapping tasks there's no state stores at all).

When you add new threads either within the same node, or on a different
node, after rebalance each thread should be processing 10 tasks, and hence
owning corresponding set of state stores due to rebalance. The new thread
will first restore the state stores it gets assigned before start
processing.


> 2) What happens to the userStream map task? Will the new node be able to
> process this task while the state store is rebuilding/reloading? My reading
> of the code suggests that this map process will be paused on the new node
> while the state store is rebuilt. The effect of this is that it will lead
> to a delay in events reaching the original node's partitions, which will be
> seen as late-arriving events. Am I right in this assessment?
>
>
Currently the thread will NOT start processing any tasks until ALL stateful
tasks completes restoring (stateless tasks, like the map tasks in your
example never needs restoration at all). There's an open JIRA for making it
customizable but I cannot find it currently.


> 3) How does scaling up work with standby state-store replicas? From my
> reading of the code, it appears that scaling a node up will result in a
> reabalance, with the state assigned to the new node being rebuilt first
> (leading to a pause in processing). Following this, the standy replicas are
> populated. Am I correct in this reading?
>
> Standby tasks are running in parallel with active stream tasks, and it
simply reads from the changelog topic in read time and populate the standby
store replica; when scaling out, the instances with standby tasks will be
preferred over those who do not have any standby for the task, and hence
when restoring only a very small amount of data needs to be restored
(think: the standby replica of the store is already populated up to offset
90 at the rebalance, while the active task is writing to the changelog
topic with log end offset 100, so you only need to restore 90 - 100 instead
of 0 - 100).


> 4) If my reading in #3 is correct, would it be possible to pre-populate the
> standby stores on scale-up before initiating active-task transfer? This
> would allow seamless scale-up and scale-down without requiring any pauses
> for rebuilding state. I am interested in kicking this off as a KIP if so,
> but would appreciate any JIRAs or related KIPs to read up on prior to
> digging into this.
>
> Yes, there is some discussions about this here:
https://issues.apache.org/jira/browse/KAFKA-6145


>
> Thanks
>
> Adam Bellemare
>


-- 
-- Guozhang

Build failed in Jenkins: kafka-2.0-jdk8 #228

[Apache Jenkins Server]

See 


Changes:

[github] MINOR: Add check all topics created check streams broker bounce test

--
[...truncated 436.02 KB...]
kafka.server.ClientQuotaManagerTest > testUserQuotaParsing STARTED

kafka.server.ClientQuotaManagerTest > testUserQuotaParsing PASSED

kafka.server.ClientQuotaManagerTest > testClientIdQuotaParsing STARTED

kafka.server.ClientQuotaManagerTest > testClientIdQuotaParsing PASSED

kafka.server.ClientQuotaManagerTest > testQuotaViolation STARTED

kafka.server.ClientQuotaManagerTest > testQuotaViolation PASSED

kafka.server.ClientQuotaManagerTest > testRequestPercentageQuotaViolation 
STARTED

kafka.server.ClientQuotaManagerTest > testRequestPercentageQuotaViolation PASSED

kafka.server.ClientQuotaManagerTest > testQuotaConfigPrecedence STARTED

kafka.server.ClientQuotaManagerTest > testQuotaConfigPrecedence PASSED

kafka.server.ClientQuotaManagerTest > testExpireQuotaSensors STARTED

kafka.server.ClientQuotaManagerTest > testExpireQuotaSensors PASSED

kafka.server.ClientQuotaManagerTest > testClientIdNotSanitized STARTED

kafka.server.ClientQuotaManagerTest > testClientIdNotSanitized PASSED

kafka.server.ClientQuotaManagerTest > testExpireThrottleTimeSensor STARTED

kafka.server.ClientQuotaManagerTest > testExpireThrottleTimeSensor PASSED

kafka.server.ClientQuotaManagerTest > testUserClientIdQuotaParsing STARTED

kafka.server.ClientQuotaManagerTest > testUserClientIdQuotaParsing PASSED

kafka.server.ClientQuotaManagerTest > 
testUserClientQuotaParsingIdWithDefaultClientIdQuota STARTED

kafka.server.ClientQuotaManagerTest > 
testUserClientQuotaParsingIdWithDefaultClientIdQuota PASSED

kafka.server.ReplicaManagerQuotasTest > shouldGetBothMessagesIfQuotasAllow 
STARTED

kafka.server.ReplicaManagerQuotasTest > shouldGetBothMessagesIfQuotasAllow 
PASSED

kafka.server.ReplicaManagerQuotasTest > 
testCompleteInDelayedFetchWithReplicaThrottling STARTED

kafka.server.ReplicaManagerQuotasTest > 
testCompleteInDelayedFetchWithReplicaThrottling PASSED

kafka.server.ReplicaManagerQuotasTest > 
shouldExcludeSubsequentThrottledPartitions STARTED

kafka.server.ReplicaManagerQuotasTest > 
shouldExcludeSubsequentThrottledPartitions PASSED

kafka.server.ReplicaManagerQuotasTest > 
shouldGetNoMessagesIfQuotasExceededOnSubsequentPartitions STARTED

kafka.server.ReplicaManagerQuotasTest > 
shouldGetNoMessagesIfQuotasExceededOnSubsequentPartitions PASSED

kafka.server.ReplicaManagerQuotasTest > shouldIncludeInSyncThrottledReplicas 
STARTED

kafka.server.ReplicaManagerQuotasTest > shouldIncludeInSyncThrottledReplicas 
PASSED

kafka.server.DynamicBrokerConfigTest > testPasswordConfigEncryption STARTED

kafka.server.DynamicBrokerConfigTest > testPasswordConfigEncryption PASSED

kafka.server.DynamicBrokerConfigTest > testSecurityConfigs STARTED

kafka.server.DynamicBrokerConfigTest > testSecurityConfigs PASSED

kafka.server.DynamicBrokerConfigTest > testSynonyms STARTED

kafka.server.DynamicBrokerConfigTest > testSynonyms PASSED

kafka.server.DynamicBrokerConfigTest > 
testDynamicConfigInitializationWithoutConfigsInZK STARTED

kafka.server.DynamicBrokerConfigTest > 
testDynamicConfigInitializationWithoutConfigsInZK PASSED

kafka.server.DynamicBrokerConfigTest > testConfigUpdateWithSomeInvalidConfigs 
STARTED

kafka.server.DynamicBrokerConfigTest > testConfigUpdateWithSomeInvalidConfigs 
PASSED

kafka.server.DynamicBrokerConfigTest > testDynamicListenerConfig STARTED

kafka.server.DynamicBrokerConfigTest > testDynamicListenerConfig PASSED

kafka.server.DynamicBrokerConfigTest > testReconfigurableValidation STARTED

kafka.server.DynamicBrokerConfigTest > testReconfigurableValidation PASSED

kafka.server.DynamicBrokerConfigTest > testConfigUpdate STARTED

kafka.server.DynamicBrokerConfigTest > testConfigUpdate PASSED

kafka.server.DynamicBrokerConfigTest > testPasswordConfigEncoderSecretChange 
STARTED

kafka.server.DynamicBrokerConfigTest > testPasswordConfigEncoderSecretChange 
PASSED

kafka.server.DynamicBrokerConfigTest > 
testConfigUpdateWithReconfigurableValidationFailure STARTED

kafka.server.DynamicBrokerConfigTest > 
testConfigUpdateWithReconfigurableValidationFailure PASSED

kafka.server.ThrottledChannelExpirationTest > testThrottledChannelDelay STARTED

kafka.server.ThrottledChannelExpirationTest > testThrottledChannelDelay PASSED

kafka.server.ThrottledChannelExpirationTest > 
testCallbackInvocationAfterExpiration STARTED

kafka.server.ThrottledChannelExpirationTest > 
testCallbackInvocationAfterExpiration PASSED

kafka.server.SaslApiVersionsRequestTest > 
testApiVersionsRequestWithUnsupportedVersion STARTED

kafka.server.SaslApiVersionsRequestTest > 
testApiVersionsRequestWithUnsupportedVersion PASSED

kafka.server.SaslApiVersionsRequestTest > 
testApiVersionsRequestBeforeSaslHandshakeRequest STARTED

kafka.server.SaslApiVersionsRequestTest > 

Build failed in Jenkins: kafka-2.1-jdk8 #133

[Apache Jenkins Server]

See 


Changes:

[rajinisivaram] KAFKA-7945: Calc refresh time correctly when token created in 
the past

--
[...truncated 701.42 KB...]
kafka.utils.json.JsonValueTest > testDecodeInt PASSED

kafka.utils.json.JsonValueTest > testDecodeMap STARTED

kafka.utils.json.JsonValueTest > testDecodeMap PASSED

kafka.utils.json.JsonValueTest > testDecodeSeq STARTED

kafka.utils.json.JsonValueTest > testDecodeSeq PASSED

kafka.utils.json.JsonValueTest > testJsonObjectGet STARTED

kafka.utils.json.JsonValueTest > testJsonObjectGet PASSED

kafka.utils.json.JsonValueTest > testJsonValueEquals STARTED

kafka.utils.json.JsonValueTest > testJsonValueEquals PASSED

kafka.utils.json.JsonValueTest > testJsonArrayIterator STARTED

kafka.utils.json.JsonValueTest > testJsonArrayIterator PASSED

kafka.utils.json.JsonValueTest > testJsonObjectApply STARTED

kafka.utils.json.JsonValueTest > testJsonObjectApply PASSED

kafka.utils.json.JsonValueTest > testDecodeBoolean STARTED

kafka.utils.json.JsonValueTest > testDecodeBoolean PASSED

kafka.utils.TopicFilterTest > testWhitelists STARTED

kafka.utils.TopicFilterTest > testWhitelists PASSED

kafka.utils.PasswordEncoderTest > testEncoderConfigChange STARTED

kafka.utils.PasswordEncoderTest > testEncoderConfigChange PASSED

kafka.utils.PasswordEncoderTest > testEncodeDecodeAlgorithms STARTED

kafka.utils.PasswordEncoderTest > testEncodeDecodeAlgorithms PASSED

kafka.utils.PasswordEncoderTest > testEncodeDecode STARTED

kafka.utils.PasswordEncoderTest > testEncodeDecode PASSED

unit.kafka.utils.ThrottlerTest > testThrottleDesiredRate STARTED

unit.kafka.utils.ThrottlerTest > testThrottleDesiredRate PASSED

2158 tests completed, 2 failed, 4 skipped

> Task :kafka-2.1-jdk8:core:test FAILED
> Task :testScala_2_12 FAILED

> Task :kafka-2.1-jdk8:streams:streams-scala:test

org.apache.kafka.streams.scala.StreamToTableJoinScalaIntegrationTestImplicitSerdes
 > testShouldCountClicksPerRegionWithNamedRepartitionTopic STARTED

org.apache.kafka.streams.scala.StreamToTableJoinScalaIntegrationTestImplicitSerdes
 > testShouldCountClicksPerRegionWithNamedRepartitionTopic FAILED
java.lang.NoSuchMethodError: 
kafka.utils.Logging.$init$(Lkafka/utils/Logging;)V

org.apache.kafka.streams.scala.StreamToTableJoinScalaIntegrationTestImplicitSerdes
 > testShouldCountClicksPerRegionJava STARTED

org.apache.kafka.streams.scala.StreamToTableJoinScalaIntegrationTestImplicitSerdes
 > testShouldCountClicksPerRegionJava FAILED
java.lang.NoSuchMethodError: 
kafka.utils.Logging.$init$(Lkafka/utils/Logging;)V

org.apache.kafka.streams.scala.StreamToTableJoinScalaIntegrationTestImplicitSerdes
 > testShouldCountClicksPerRegion STARTED

org.apache.kafka.streams.scala.StreamToTableJoinScalaIntegrationTestImplicitSerdes
 > testShouldCountClicksPerRegion FAILED
java.lang.NoSuchMethodError: 
kafka.utils.Logging.$init$(Lkafka/utils/Logging;)V

org.apache.kafka.streams.scala.TopologyTest > 
shouldBuildIdenticalTopologyInJavaNScalaJoin STARTED

org.apache.kafka.streams.scala.TopologyTest > 
shouldBuildIdenticalTopologyInJavaNScalaJoin PASSED

org.apache.kafka.streams.scala.TopologyTest > 
shouldBuildIdenticalTopologyInJavaNScalaSimple STARTED

org.apache.kafka.streams.scala.TopologyTest > 
shouldBuildIdenticalTopologyInJavaNScalaSimple PASSED

org.apache.kafka.streams.scala.TopologyTest > 
shouldBuildIdenticalTopologyInJavaNScalaAggregate STARTED

org.apache.kafka.streams.scala.TopologyTest > 
shouldBuildIdenticalTopologyInJavaNScalaAggregate PASSED

org.apache.kafka.streams.scala.TopologyTest > 
shouldBuildIdenticalTopologyInJavaNScalaTransform STARTED

org.apache.kafka.streams.scala.TopologyTest > 
shouldBuildIdenticalTopologyInJavaNScalaTransform PASSED

org.apache.kafka.streams.scala.WordCountTest > testShouldCountWordsMaterialized 
STARTED

org.apache.kafka.streams.scala.WordCountTest > testShouldCountWordsMaterialized 
FAILED
java.lang.NoSuchMethodError: 
kafka.utils.Logging.$init$(Lkafka/utils/Logging;)V

org.apache.kafka.streams.scala.WordCountTest > testShouldCountWordsJava STARTED

org.apache.kafka.streams.scala.WordCountTest > testShouldCountWordsJava FAILED
java.lang.NoSuchMethodError: 
kafka.utils.Logging.$init$(Lkafka/utils/Logging;)V

org.apache.kafka.streams.scala.WordCountTest > testShouldCountWords STARTED

org.apache.kafka.streams.scala.WordCountTest > testShouldCountWords FAILED
java.lang.NoSuchMethodError: 
kafka.utils.Logging.$init$(Lkafka/utils/Logging;)V

org.apache.kafka.streams.scala.kstream.MaterializedTest > Create a Materialized 
should create a Materialized with Serdes STARTED

org.apache.kafka.streams.scala.kstream.MaterializedTest > Create a Materialized 
should create a Materialized with Serdes PASSED

org.apache.kafka.streams.scala.kstream.MaterializedTest > Create a Materialize 
with a store name should create a Materialized with Serdes and a store name 

[jira] [Created] (KAFKA-7966) Flaky Test DynamicBrokerReconfigurationTest#testLogCleanerConfig

[Matthias J. Sax (JIRA)]

Matthias J. Sax created KAFKA-7966:
--

 Summary: Flaky Test 
DynamicBrokerReconfigurationTest#testLogCleanerConfig
 Key: KAFKA-7966
 URL: https://issues.apache.org/jira/browse/KAFKA-7966
 Project: Kafka
  Issue Type: Bug
  Components: core, unit tests
Affects Versions: 2.2.0
Reporter: Matthias J. Sax
 Fix For: 2.2.0


To get stable nightly builds for `2.2` release, I create tickets for all 
observed test failures.

[https://jenkins.confluent.io/job/apache-kafka-test/job/2.2/22/]
{quote}java.lang.AssertionError: Partition [__consumer_offsets,0] metadata not 
propagated after 15000 ms at kafka.utils.TestUtils$.fail(TestUtils.scala:356) 
at kafka.utils.TestUtils$.waitUntilTrue(TestUtils.scala:766) at 
kafka.utils.TestUtils$.waitUntilMetadataIsPropagated(TestUtils.scala:855) at 
kafka.utils.TestUtils$.$anonfun$createTopic$1(TestUtils.scala:303) at 
kafka.utils.TestUtils$.$anonfun$createTopic$1$adapted(TestUtils.scala:302) at 
scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:237) at 
scala.collection.immutable.Range.foreach(Range.scala:158) at 
scala.collection.TraversableLike.map(TraversableLike.scala:237) at 
scala.collection.TraversableLike.map$(TraversableLike.scala:230) at 
scala.collection.AbstractTraversable.map(Traversable.scala:108) at 
kafka.utils.TestUtils$.createTopic(TestUtils.scala:302) at 
kafka.server.DynamicBrokerReconfigurationTest.setUp(DynamicBrokerReconfigurationTest.scala:137){quote}



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (KAFKA-7965) Flaky Test ConsumerBounceTest#testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup

[Matthias J. Sax (JIRA)]

Matthias J. Sax created KAFKA-7965:
--

 Summary: Flaky Test 
ConsumerBounceTest#testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup
 Key: KAFKA-7965
 URL: https://issues.apache.org/jira/browse/KAFKA-7965
 Project: Kafka
  Issue Type: Bug
Reporter: Matthias J. Sax


To get stable nightly builds for `2.2` release, I create tickets for all 
observed test failures.

[https://jenkins.confluent.io/job/apache-kafka-test/job/2.2/21/]
{quote}java.lang.AssertionError: Received 0, expected at least 68 at 
org.junit.Assert.fail(Assert.java:88) at 
org.junit.Assert.assertTrue(Assert.java:41) at 
kafka.api.ConsumerBounceTest.receiveAndCommit(ConsumerBounceTest.scala:557) at 
kafka.api.ConsumerBounceTest.$anonfun$testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup$1(ConsumerBounceTest.scala:320)
 at 
kafka.api.ConsumerBounceTest.$anonfun$testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup$1$adapted(ConsumerBounceTest.scala:319)
 at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62) at 
scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55) at 
scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49) at 
kafka.api.ConsumerBounceTest.testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup(ConsumerBounceTest.scala:319){quote}



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (KAFKA-7964) Flaky Test ConsumerBounceTest#testConsumerReceivesFatalExceptionWhenGroupPassesMaxSize

[Matthias J. Sax (JIRA)]

Matthias J. Sax created KAFKA-7964:
--

 Summary: Flaky Test 
ConsumerBounceTest#testConsumerReceivesFatalExceptionWhenGroupPassesMaxSize
 Key: KAFKA-7964
 URL: https://issues.apache.org/jira/browse/KAFKA-7964
 Project: Kafka
  Issue Type: Bug
  Components: clients, unit tests
Affects Versions: 2.2.0
Reporter: Matthias J. Sax
 Fix For: 2.2.0


To get stable nightly builds for `2.2` release, I create tickets for all 
observed test failures.

[https://jenkins.confluent.io/job/apache-kafka-test/job/2.2/21/]
{quote}java.lang.AssertionError: expected:<100> but was:<0> at 
org.junit.Assert.fail(Assert.java:88) at 
org.junit.Assert.failNotEquals(Assert.java:834) at 
org.junit.Assert.assertEquals(Assert.java:645) at 
org.junit.Assert.assertEquals(Assert.java:631) at 
kafka.api.ConsumerBounceTest.receiveExactRecords(ConsumerBounceTest.scala:551) 
at 
kafka.api.ConsumerBounceTest.$anonfun$testConsumerReceivesFatalExceptionWhenGroupPassesMaxSize$2(ConsumerBounceTest.scala:409)
 at 
kafka.api.ConsumerBounceTest.$anonfun$testConsumerReceivesFatalExceptionWhenGroupPassesMaxSize$2$adapted(ConsumerBounceTest.scala:408)
 at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62) at 
scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55) at 
scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49) at 
kafka.api.ConsumerBounceTest.testConsumerReceivesFatalExceptionWhenGroupPassesMaxSize(ConsumerBounceTest.scala:408){quote}



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Build failed in Jenkins: kafka-2.2-jdk8 #25

[Apache Jenkins Server]

See 


Changes:

[jason] KAFKA-7920; Do not permit zstd produce requests until IBP is updated to

[github] MINOR: Add all topics created check streams broker bounce test (2.2)

--
[...truncated 2.50 MB...]

kafka.controller.PartitionStateMachineTest > 
testUpdatingOfflinePartitionsCountDuringTopicDeletion PASSED

kafka.controller.PartitionStateMachineTest > 
testOfflinePartitionToOnlinePartitionTransitionErrorCodeFromStateLookup STARTED

kafka.controller.PartitionStateMachineTest > 
testOfflinePartitionToOnlinePartitionTransitionErrorCodeFromStateLookup PASSED

kafka.controller.PartitionStateMachineTest > 
testOnlinePartitionToOnlineTransitionForControlledShutdown STARTED

kafka.controller.PartitionStateMachineTest > 
testOnlinePartitionToOnlineTransitionForControlledShutdown PASSED

kafka.controller.PartitionStateMachineTest > 
testOfflinePartitionToOnlinePartitionTransitionZkUtilsExceptionFromStateLookup 
STARTED

kafka.controller.PartitionStateMachineTest > 
testOfflinePartitionToOnlinePartitionTransitionZkUtilsExceptionFromStateLookup 
PASSED

kafka.controller.PartitionStateMachineTest > 
testNoOfflinePartitionsChangeForTopicsBeingDeleted STARTED

kafka.controller.PartitionStateMachineTest > 
testNoOfflinePartitionsChangeForTopicsBeingDeleted PASSED

kafka.controller.PartitionStateMachineTest > 
testInvalidOnlinePartitionToNonexistentPartitionTransition STARTED

kafka.controller.PartitionStateMachineTest > 
testInvalidOnlinePartitionToNonexistentPartitionTransition PASSED

kafka.controller.PartitionStateMachineTest > 
testInvalidOfflinePartitionToNewPartitionTransition STARTED

kafka.controller.PartitionStateMachineTest > 
testInvalidOfflinePartitionToNewPartitionTransition PASSED

kafka.controller.PartitionStateMachineTest > 
testOfflinePartitionToOnlinePartitionTransition STARTED

kafka.controller.PartitionStateMachineTest > 
testOfflinePartitionToOnlinePartitionTransition PASSED

kafka.controller.ControllerEventManagerTest > testEventThatThrowsException 
STARTED

kafka.controller.ControllerEventManagerTest > testEventThatThrowsException 
PASSED

kafka.controller.ControllerEventManagerTest > testSuccessfulEvent STARTED

kafka.controller.ControllerEventManagerTest > testSuccessfulEvent PASSED

kafka.controller.ControllerFailoverTest > testHandleIllegalStateException 
STARTED

kafka.controller.ControllerFailoverTest > testHandleIllegalStateException PASSED

kafka.network.SocketServerTest > testGracefulClose STARTED

kafka.network.SocketServerTest > testGracefulClose PASSED

kafka.network.SocketServerTest > 
testSendActionResponseWithThrottledChannelWhereThrottlingAlreadyDone STARTED

kafka.network.SocketServerTest > 
testSendActionResponseWithThrottledChannelWhereThrottlingAlreadyDone PASSED

kafka.network.SocketServerTest > controlThrowable STARTED

kafka.network.SocketServerTest > controlThrowable PASSED

kafka.network.SocketServerTest > testRequestMetricsAfterStop STARTED

kafka.network.SocketServerTest > testRequestMetricsAfterStop PASSED

kafka.network.SocketServerTest > testConnectionIdReuse STARTED

kafka.network.SocketServerTest > testConnectionIdReuse PASSED

kafka.network.SocketServerTest > testClientDisconnectionUpdatesRequestMetrics 
STARTED

kafka.network.SocketServerTest > testClientDisconnectionUpdatesRequestMetrics 
PASSED

kafka.network.SocketServerTest > testProcessorMetricsTags STARTED

kafka.network.SocketServerTest > testProcessorMetricsTags PASSED

kafka.network.SocketServerTest > testMaxConnectionsPerIp STARTED

kafka.network.SocketServerTest > testMaxConnectionsPerIp PASSED

kafka.network.SocketServerTest > testConnectionId STARTED

kafka.network.SocketServerTest > testConnectionId PASSED

kafka.network.SocketServerTest > 
testBrokerSendAfterChannelClosedUpdatesRequestMetrics STARTED

kafka.network.SocketServerTest > 
testBrokerSendAfterChannelClosedUpdatesRequestMetrics PASSED

kafka.network.SocketServerTest > testNoOpAction STARTED

kafka.network.SocketServerTest > testNoOpAction PASSED

kafka.network.SocketServerTest > simpleRequest STARTED

kafka.network.SocketServerTest > simpleRequest PASSED

kafka.network.SocketServerTest > closingChannelException STARTED

kafka.network.SocketServerTest > closingChannelException PASSED

kafka.network.SocketServerTest > 
testSendActionResponseWithThrottledChannelWhereThrottlingInProgress STARTED

kafka.network.SocketServerTest > 
testSendActionResponseWithThrottledChannelWhereThrottlingInProgress PASSED

kafka.network.SocketServerTest > testIdleConnection STARTED

kafka.network.SocketServerTest > testIdleConnection PASSED

kafka.network.SocketServerTest > 
testClientDisconnectionWithStagedReceivesFullyProcessed STARTED

kafka.network.SocketServerTest > 
testClientDisconnectionWithStagedReceivesFullyProcessed PASSED

kafka.network.SocketServerTest > testZeroMaxConnectionsPerIp STARTED

kafka.network.SocketServerTest > 

Re: [VOTE] KIP-382 MirrorMaker 2.0

[Ryanne Dolan]

Hey y'all, I'm happy to announce that I've created a PR for MM2:

https://github.com/apache/kafka/pull/6295

Please take a look!

Ryanne


On Tue, Jan 22, 2019 at 11:43 AM Ryanne Dolan  wrote:

> Thanks all, this is a large KIP and has sparked a lot of great
> discussion, both on and off the dev list. I'm closing the vote with
> the following results:
>
> +12 non-binding
> +3 binding
>
> The KIP is accepted!
>
> Thanks,
> Ryanne
>
> On Fri, Jan 11, 2019 at 9:48 PM Becket Qin  wrote:
> >
> > Hi Ryanne,
> >
> > Thanks for the KIP and patient discussion. +1 from me as well.
> >
> > Jiangjie (Becket) Qin
> >
> > On Fri, Jan 11, 2019 at 1:11 AM Jun Rao  wrote:
> >>
> >> Hi, Ryanne,
> >>
> >> Thanks for the explanation. All make sense to me now. +1 on the KIP
> from me.
> >>
> >> Jun
> >>
> >> On Wed, Jan 9, 2019 at 7:16 PM Ryanne Dolan 
> wrote:
> >>
> >> > Thanks Jun.
> >> >
> >> > > 103. My point was that the MirrorMakerConnector can die while the
> >> > Heartbeat connector is still alive. So, one can't solely rely on
> Heartbeat
> >> > for monitoring?
> >> >
> >> > Each cluster will have a heartbeat topic produced by
> >> > MirrorHeartbeatConnector, which doesn't have an associated "source"
> other
> >> > than time. This topic gets picked up by downstream
> MirrorSourceConnectors
> >> > and replicated like A.heartbeat. So the heartbeat topic itself isn't
> >> > particular useful for monitoring, but the downstream A.heartbeat
> shows that
> >> > heartbeats are being replicated successfully from A -> B. If a
> >> > MirrorSourceConnector fails while replicating A -> B, you'd still see
> >> > heartbeats in cluster B, but not A.heartbeat.
> >> >
> >> > 105. You're correct, you don't need to know "B" in order to go from
> A's
> >> > topic1 to B's A.topic1, i.e. migrating downstream. But you need to
> know "B"
> >> > to go from A's B.topic1 to B's topic1. In the latter case, you are
> >> > consuming a remote topic to begin with, and then migrating to the
> source
> >> > cluster, i.e. migrating upstream. N.B. you strip the "B" prefix in
> this
> >> > case, rather than add the "A" prefix. And you can't just strip all
> >> > prefixes, because you could be migrating from e.g. A's C.topic1 to B's
> >> > C.topic1, i.e. migrating "laterally", if you will.
> >> >
> >> > I suppose we could break this out into multiple methods (upstream,
> >> > downstream, lateral etc), but I think that would add a lot more
> complexity
> >> > and confusion to the API. By providing both A and B, the single
> method can
> >> > always figure out what to do.
> >> >
> >> > 107. done
> >> >
> >> > Thanks,
> >> > Ryanne
> >> >
> >> >
> >> >
> >> >
> >> > On Wed, Jan 9, 2019 at 6:11 PM Jun Rao  wrote:
> >> >
> >> >> Hi, Ryanne,
> >> >>
> >> >> 103. My point was that the MirrorMakerConnector can die while the
> >> >> Heartbeat connector is still alive. So, one can't solely rely on
> Heartbeat
> >> >> for monitoring?
> >> >>
> >> >> 105. Hmm, maybe I don't understand how this is done. Let's say we
> replica
> >> >> topic1 from cluster A to cluster B. My understanding is that to
> translate
> >> >> the offset from A to B for a consumer group, we read A.checkpoint
> file in
> >> >> cluster B to get the timestamp of the last checkpointed offset, call
> >> >> consumer.offsetsForTimes() on A.topic1 in cluster B to translate the
> >> >> timestamp to a local offset, and return  offset>.
> >> >> Is that right? If so, in all steps, we don't need to know the
> >> >> targetClusterAlias B. We just need to know the connection string to
> >> >> cluster B, which targetConsumerConfig provides.
> >> >>
> >> >> 107. Thanks. Could you add that description to the KIP?
> >> >>
> >> >> Thanks,
> >> >>
> >> >> Jun
> >> >>
> >> >> On Mon, Jan 7, 2019 at 3:50 PM Ryanne Dolan 
> >> >> wrote:
> >> >>
> >> >>> Thanks Jun, I've updated the KIP as requested. Brief notes below:
> >> >>>
> >> >>> 100. added "...out-of-the-box (without custom handlers)..."
> >> >>>
> >> >>> 101. done. Good idea to include a MessageFormatter.
> >> >>>
> >> >>> 102. done.
> >> >>>
> >> >>> > 103. [...] why is Heartbeat a separate connector?
> >> >>>
> >> >>> Heartbeats themselves are replicated via
> MirrorSource/SinkConnector, so
> >> >>> if replication stops, you'll stop seeing heartbeats in downstream
> clusters.
> >> >>> I've updated the KIP to make this clearer and have added a bullet to
> >> >>> Rejected Alternatives.
> >> >>>
> >> >>> 104. added "heartbeat.retention.ms", "checkpoint.retention.ms",
> thanks.
> >> >>> The heartbeat topic doesn't need to be compacted.
> >> >>>
> >> >>> > 105. [...] I am not sure why targetClusterAlias is useful
> >> >>>
> >> >>> In order to map A's B.topic1 to B's topic1, we need to know B.
> >> >>>
> >> >>> > 106. [...] should the following properties be prefixed with
> "consumer."
> >> >>>
> >> >>> No, they are part of Connect's worker config.
> >> >>>
> >> >>> > 107. So, essentially it's running multiple logical connect
> clusters on
> >> >>> the same shared 

[jira] [Resolved] (KAFKA-7945) ExpiringCredentialRefreshingLogin - timeout value is negative

[Rajini Sivaram (JIRA)]

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

Rajini Sivaram resolved KAFKA-7945.
---
   Resolution: Fixed
 Reviewer: Rajini Sivaram
Fix Version/s: 2.1.2
   2.2.0

> ExpiringCredentialRefreshingLogin - timeout value is negative
> -
>
> Key: KAFKA-7945
> URL: https://issues.apache.org/jira/browse/KAFKA-7945
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 2.0.1
>Reporter: Denis Ogun
>Assignee: Ron Dagostino
>Priority: Major
> Fix For: 2.2.0, 2.1.2
>
>
> There was an issue with one of our Kafka consumers no longer sending a valid 
> OAuth token. Looking at the logs, there seems to be an error in some of the 
> math in the timestamp calculation:
>  
> {code:java}
> 14 Feb 2019 06:42:45,694 Expiring credential expires at 
> 2019-02-14T06:48:21.000+, so buffer times of 60 and 300 seconds at the 
> front and back, respectively, cannot be accommodated. We will refresh at 
> 2019-02-14T06:01:39.078+.
> 14 Feb 2019 06:42:45,694 org.apache.kafka.common.utils.KafkaThread: Uncaught 
> exception in thread
> java.lang.IllegalArgumentException: timeout value is negative
> at java.lang.Thread.sleep(Native Method) ~[?:1.8.0_202]
> at org.apache.kafka.common.utils.SystemTime.sleep(SystemTime.java:45) 
> ~[kafka-clients-2.x.jar:?]
> at 
> org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin$Refresher.run(ExpiringCredentialRefreshingLogin.java:86)
>  ~[kafka-clients-2.x.jar:?]
> at java.lang.Thread.run(Thread.java:748) [?:1.8.0_202]{code}
>  
> At this point the refresh logic would never recover and so the producer 
> couldn't consume until we restarted the process.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Jenkins build is back to normal : kafka-trunk-jdk8 #3399

[Apache Jenkins Server]

See 

Re: [DISCUSS] KIP-411: Add option to make Kafka Connect task client ID values unique

[Ryanne Dolan]

Thanks Paul, this is great. This will make monitoring Connect a ton easier.

Ryanne

On Wed, Feb 20, 2019 at 1:24 PM Paul Davidson
 wrote:

> I have updated KIP-411 to propose changing the default client id - see:
>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-411%3A+Make+default+Kafka+Connect+worker+task+client+IDs+distinct
>
>
> There is also an PR ready to go here:
> https://github.com/apache/kafka/pull/6097
>
> On Fri, Jan 11, 2019 at 3:39 PM Paul Davidson 
> wrote:
>
> > Hi everyone.  We seem to have agreement that the ideal approach is to
> > alter the default client ids. Now I'm wondering about the best process to
> > proceed. Will the change in default behaviour require a new KIP, given it
> > will affect existing deployments?  Would I be best to repurpose this
> > KIP-411, or am I best to  create a new KIP? Thanks!
> >
> > Paul
> >
> > On Tue, Jan 8, 2019 at 7:16 PM Randall Hauch  wrote:
> >
> >> Hi, Paul.
> >>
> >> I concur with the others, and I like the new approach that avoids a new
> >> configuration, especially because it does not change the behavior for
> >> anyone already using `producer.client.id` and/or `consumer.client.id`.
> I
> >> did leave a few comments on the PR. Perhaps the biggest one is whether
> the
> >> producer used for the sink task error reporter (for dead letter queue)
> >> should be `connector-producer-`, and whether that is
> >> distinct
> >> enough from source tasks, which will be of the form
> >> `connector-producer-`. Maybe it is fine. (The other
> >> comments were minor.)
> >>
> >> Best regards,
> >>
> >> Randall
> >>
> >> On Mon, Jan 7, 2019 at 1:19 PM Paul Davidson 
> >> wrote:
> >>
> >> > Thanks all. I've submitted a new PR with a possible implementation:
> >> > https://github.com/apache/kafka/pull/6097. Note I did not include the
> >> > group
> >> > ID as part of the default client ID, mainly to avoid the connector
> name
> >> > appearing twice by default. As noted in the original Jira (
> >> > https://issues.apache.org/jira/browse/KAFKA-5061), leaving out the
> >> group
> >> > ID
> >> > could lead to naming conflicts if multiple clusters run the same Kafka
> >> > cluster. This would probably not be a problem for many (including us)
> as
> >> > metrics exporters can usually be configured to include a cluster ID
> and
> >> > guarantee uniqueness. Will be interested to hear your thoughts on
> this.
> >> >
> >> > Paul
> >> >
> >> >
> >> >
> >> > On Mon, Jan 7, 2019 at 10:27 AM Ryanne Dolan 
> >> > wrote:
> >> >
> >> > > I'd also prefer to avoid the new configuration property if possible.
> >> > Seems
> >> > > like a lighter touch without it.
> >> > >
> >> > > Ryanne
> >> > >
> >> > > On Sun, Jan 6, 2019 at 7:25 PM Paul Davidson <
> >> pdavid...@salesforce.com>
> >> > > wrote:
> >> > >
> >> > > > Hi Konstantine,
> >> > > >
> >> > > > Thanks for your feedback!  I think my reply to Ewen covers most of
> >> your
> >> > > > points, and I mostly agree.  If there is general agreement that
> >> > changing
> >> > > > the default behavior is preferable to a config change I will
> update
> >> my
> >> > PR
> >> > > > to use  that approach.
> >> > > >
> >> > > > Paul
> >> > > >
> >> > > > On Fri, Jan 4, 2019 at 5:55 PM Konstantine Karantasis <
> >> > > > konstant...@confluent.io> wrote:
> >> > > >
> >> > > > > Hi Paul.
> >> > > > >
> >> > > > > I second Ewen and I intended to give similar feedback:
> >> > > > >
> >> > > > > 1) Can we avoid a config altogether?
> >> > > > > 2) If we prefer to add a config anyways, can we use a set of
> >> allowed
> >> > > > values
> >> > > > > instead of a boolean, even if initially these values are only
> >> two? As
> >> > > the
> >> > > > > discussion on Jira highlights, there is a potential for more
> >> naming
> >> > > > > conventions in the future, even if now the extra functionality
> >> > doesn't
> >> > > > seem
> >> > > > > essential. It's not optimal to have to deprecate a config
> instead
> >> of
> >> > > just
> >> > > > > extending its set of values.
> >> > > > > 3) I agree, the config name sounds too general. How about
> >> > > > > "client.ids.naming.policy" or "client.ids.naming" if you want
> two
> >> > more
> >> > > > > options?
> >> > > > >
> >> > > > > Konstantine
> >> > > > >
> >> > > > > On Fri, Jan 4, 2019 at 7:38 AM Ewen Cheslack-Postava <
> >> > > e...@confluent.io>
> >> > > > > wrote:
> >> > > > >
> >> > > > > > Hi Paul,
> >> > > > > >
> >> > > > > > Thanks for the KIP. A few comments.
> >> > > > > >
> >> > > > > > To me, biggest question here is if we can fix this behavior
> >> without
> >> > > > > adding
> >> > > > > > a config. In particular, today, we don't even set the
> client.id
> >> > for
> >> > > > the
> >> > > > > > producer and consumer at all, right? The *only* way it is set
> >> is if
> >> > > you
> >> > > > > > include an override in the worker config, but in that case you
> >> need
> >> > > to
> >> > > > be
> >> > > > > > explicitly opting in with a `producer.` or `consumer.` prefix,
> >> i.e.
> 

Re: [DISCUSS] KIP-236 Interruptible Partition Reassignment

[George Li]

Hi,
After discussing with Tom, Harsha and I are picking up KIP-236.  The work 
focused on safely/cleanly cancel / rollback pending reassignments in a timely 
fashion.  Pull Request #6296  Still working on more integration/system tests. 
Please review and provide feedbacks/suggestions. 
Thanks,George

On Saturday, December 23, 2017, 0:51:13 GMT, Jun Rao  wrote:

Hi, Tom,

Thanks for the reply.

10. That's a good thought. Perhaps it's better to get rid of
/admin/reassignment_requests
too. The window when a controller is not available is small. So, we can
just failed the admin client if the controller is not reachable after the
timeout.

13. With the changes in 10, the old approach is handled through ZK callback
and the new approach is through Kafka RPC. The ordering between the two is
kind of arbitrary. Perhaps the ordering can just be based on the order that
the reassignment is added to the controller request queue. From there, we
can either do the overriding or the prevention.

Jun


On Fri, Dec 22, 2017 at 7:31 AM, Tom Bentley  wrote:

> Hi Jun,
>
> Thanks for responding, my replies are inline:
>
> 10. You explanation makes sense. My remaining concern is the additional ZK
> > writes in the proposal. With the proposal, we will need to do following
> > writes in ZK.
> >
> > a. write new assignment in /admin/reassignment_requests
> >
> > b. write new assignment and additional metadata in
> > /admin/reassignments/$topic/$partition
> >
> > c. write old + new assignment  in /brokers/topics/[topic]
> >
> > d. write new assignment in /brokers/topics/[topic]
> >
> > e. delete /admin/reassignments/$topic/$partition
> >
> > So, there are quite a few ZK writes. I am wondering if it's better to
> > consolidate the info in /admin/reassignments/$topic/$partition into
> > /brokers/topics/[topic].
> > For example, we can just add some new JSON fields in
> > /brokers/topics/[topic]
> > to remember the new assignment and potentially the original replica count
> > when doing step c. Those fields with then be removed in step d. That way,
> > we can get rid of step b and e, saving 2 ZK writes per partition.
> >
>
> This seems like a great idea to me.
>
> It might also be possible to get rid of the /admin/reassignment_requests
> subtree too. I've not yet published the ideas I have for the AdminClient
> API for reassigning partitions, but given the existence of such an API, the
> route to starting a reassignment would be the AdminClient, and not
> zookeeper. In that case there is no need for /admin/reassignment_requests
> at all. The only drawback that I can see is that while it's currently
> possible to trigger a reassignment even during a controller
> election/failover that would no longer be the case if all requests had to
> go via the controller.
>
>
> > 11. What you described sounds good. We could potentially optimize the
> > dropped replicas a bit more. Suppose that assignment [0,1,2] is first
> > changed to [1,2,3] and then to [2,3,4]. When initiating the second
> > assignment, we may end up dropping replica 3 and only to restart it
> again.
> > In this case, we could only drop a replica if it's not going to be added
> > back again.
> >
>
> I had missed that, thank you! I will update the proposed algorithm to
> prevent this.
>
>
> > 13. Since this is a corner case, we can either prevent or allow
> overriding
> > with old/new mechanisms. To me, it seems that allowing is simpler to
> > implement, the order in /admin/reassignment_requests determines the
> > ordering the of override, whether that's initiated by the new way or the
> > old way.
> >
>
> That makes sense except for the corner case where:
>
> * There is no current controller and
> * Writes to both the new and old znodes happen
>
> On election of the new controller, for those partitions with both a
> reassignment_request and in /admin/reassign_partitions, we have to decide
> which should win. You could use the modification time, though there are
> some very unlikely scenarios where that doesn't work properly, for example
> if both znodes have the same mtime, or the /admin/reassign_partitions was
> updated, but the assignment of the partition wasn't changed, like this:
>
> 0. /admin/reassign_partitions has my-topic/42 = [1,2,3]
> 1. Controller stops watching.
> 2. Create /admin/reassignment_requests/request_1234 to change the
> reassignment of partition my-topic/42 = [4,5,6]
> 3. Update /admin/reassign_partitions to add your-topic/12=[7,8,9]
> 4. New controller resumes
>
>
>
> > Thanks,
> >
> > Jun
> >
> > On Tue, Dec 19, 2017 at 2:43 AM, Tom Bentley 
> > wrote:
> >
> > > Hi Jun,
> > >
> > > 10. Another concern of mine is on consistency with the current pattern.
> > The
> > > > current pattern for change notification based on ZK is (1) we first
> > write
> > > > the actual value in the entity path and then write the change
> > > notification
> > > > path, and (2)  the change notification path only includes what entity
> > has
> > > > changed but not the actual 

[jira] [Created] (KAFKA-7963) Extract hard-coded strings to centralized place

[Sophie Blee-Goldman (JIRA)]

Sophie Blee-Goldman created KAFKA-7963:
--

 Summary: Extract hard-coded strings to centralized place
 Key: KAFKA-7963
 URL: https://issues.apache.org/jira/browse/KAFKA-7963
 Project: Kafka
  Issue Type: Improvement
Reporter: Sophie Blee-Goldman


Several string literals are hard-coded into the metrics, eg 
"expired-window-record-drop" and "late-record-drop" in the window bytes stores. 
These should be moved to a sensible central location, and widespread string 
literals from these metrics may be causing memory pressure.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (KAFKA-7962) StickyAssignor: throws NullPointerException during assignments if topic is deleted

[Oleg Smirnov (JIRA)]

Oleg Smirnov created KAFKA-7962:
---

 Summary: StickyAssignor: throws NullPointerException during 
assignments if topic is deleted
 Key: KAFKA-7962
 URL: https://issues.apache.org/jira/browse/KAFKA-7962
 Project: Kafka
  Issue Type: Bug
  Components: consumer
Affects Versions: 2.1.0
 Environment: 1. MacOS, com.salesforce.kafka.test.KafkaTestUtils (kind 
of embedded kafka integration tests)
2. Linux, dockerised kafka and our service
Reporter: Oleg Smirnov
 Attachments: NPE-StickyAssignor-issues.apache.log

Integration tests with  com.salesforce.kafka.test.KafkaTestUtils, local setup, 
StickyAssignor used, local topics are created / removed, one topic is created 
in the beginning of test and without unsubscribing from it - deleted.

Same happens in real environment.

 
 # have single "topic" with 1 partition
 # single consumer subscribed to this "topic" (StickyAssignor)
 # delete "topic"

=>
 * rebalance starts, topic partition(s) is revoked
 * on assignment StickyAssignor throws exception (line 223), because 
partitionsPerTopic.("topic") returns null in for loop (topic deleted - no 
partitions are present)

 

In the provided log part, tearDown() causes topic deletion, while consumer 
still running and tries to poll data from topic.

RangeAssignor works fine (revokes partition, assigns empty set).

Problem doesn't have workaround (like handle i in onPartitionsAssigned and 
remove unsubscribe topic), because everything happens before listener called.

 



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Re: [DISCUSS] KIP-411: Add option to make Kafka Connect task client ID values unique

[Paul Davidson]

I have updated KIP-411 to propose changing the default client id - see:
https://cwiki.apache.org/confluence/display/KAFKA/KIP-411%3A+Make+default+Kafka+Connect+worker+task+client+IDs+distinct


There is also an PR ready to go here:
https://github.com/apache/kafka/pull/6097

On Fri, Jan 11, 2019 at 3:39 PM Paul Davidson 
wrote:

> Hi everyone.  We seem to have agreement that the ideal approach is to
> alter the default client ids. Now I'm wondering about the best process to
> proceed. Will the change in default behaviour require a new KIP, given it
> will affect existing deployments?  Would I be best to repurpose this
> KIP-411, or am I best to  create a new KIP? Thanks!
>
> Paul
>
> On Tue, Jan 8, 2019 at 7:16 PM Randall Hauch  wrote:
>
>> Hi, Paul.
>>
>> I concur with the others, and I like the new approach that avoids a new
>> configuration, especially because it does not change the behavior for
>> anyone already using `producer.client.id` and/or `consumer.client.id`. I
>> did leave a few comments on the PR. Perhaps the biggest one is whether the
>> producer used for the sink task error reporter (for dead letter queue)
>> should be `connector-producer-`, and whether that is
>> distinct
>> enough from source tasks, which will be of the form
>> `connector-producer-`. Maybe it is fine. (The other
>> comments were minor.)
>>
>> Best regards,
>>
>> Randall
>>
>> On Mon, Jan 7, 2019 at 1:19 PM Paul Davidson 
>> wrote:
>>
>> > Thanks all. I've submitted a new PR with a possible implementation:
>> > https://github.com/apache/kafka/pull/6097. Note I did not include the
>> > group
>> > ID as part of the default client ID, mainly to avoid the connector name
>> > appearing twice by default. As noted in the original Jira (
>> > https://issues.apache.org/jira/browse/KAFKA-5061), leaving out the
>> group
>> > ID
>> > could lead to naming conflicts if multiple clusters run the same Kafka
>> > cluster. This would probably not be a problem for many (including us) as
>> > metrics exporters can usually be configured to include a cluster ID and
>> > guarantee uniqueness. Will be interested to hear your thoughts on this.
>> >
>> > Paul
>> >
>> >
>> >
>> > On Mon, Jan 7, 2019 at 10:27 AM Ryanne Dolan 
>> > wrote:
>> >
>> > > I'd also prefer to avoid the new configuration property if possible.
>> > Seems
>> > > like a lighter touch without it.
>> > >
>> > > Ryanne
>> > >
>> > > On Sun, Jan 6, 2019 at 7:25 PM Paul Davidson <
>> pdavid...@salesforce.com>
>> > > wrote:
>> > >
>> > > > Hi Konstantine,
>> > > >
>> > > > Thanks for your feedback!  I think my reply to Ewen covers most of
>> your
>> > > > points, and I mostly agree.  If there is general agreement that
>> > changing
>> > > > the default behavior is preferable to a config change I will update
>> my
>> > PR
>> > > > to use  that approach.
>> > > >
>> > > > Paul
>> > > >
>> > > > On Fri, Jan 4, 2019 at 5:55 PM Konstantine Karantasis <
>> > > > konstant...@confluent.io> wrote:
>> > > >
>> > > > > Hi Paul.
>> > > > >
>> > > > > I second Ewen and I intended to give similar feedback:
>> > > > >
>> > > > > 1) Can we avoid a config altogether?
>> > > > > 2) If we prefer to add a config anyways, can we use a set of
>> allowed
>> > > > values
>> > > > > instead of a boolean, even if initially these values are only
>> two? As
>> > > the
>> > > > > discussion on Jira highlights, there is a potential for more
>> naming
>> > > > > conventions in the future, even if now the extra functionality
>> > doesn't
>> > > > seem
>> > > > > essential. It's not optimal to have to deprecate a config instead
>> of
>> > > just
>> > > > > extending its set of values.
>> > > > > 3) I agree, the config name sounds too general. How about
>> > > > > "client.ids.naming.policy" or "client.ids.naming" if you want two
>> > more
>> > > > > options?
>> > > > >
>> > > > > Konstantine
>> > > > >
>> > > > > On Fri, Jan 4, 2019 at 7:38 AM Ewen Cheslack-Postava <
>> > > e...@confluent.io>
>> > > > > wrote:
>> > > > >
>> > > > > > Hi Paul,
>> > > > > >
>> > > > > > Thanks for the KIP. A few comments.
>> > > > > >
>> > > > > > To me, biggest question here is if we can fix this behavior
>> without
>> > > > > adding
>> > > > > > a config. In particular, today, we don't even set the client.id
>> > for
>> > > > the
>> > > > > > producer and consumer at all, right? The *only* way it is set
>> is if
>> > > you
>> > > > > > include an override in the worker config, but in that case you
>> need
>> > > to
>> > > > be
>> > > > > > explicitly opting in with a `producer.` or `consumer.` prefix,
>> i.e.
>> > > the
>> > > > > > settings are `producer.client.id` and `consumer.client.id`.
>> > > > Otherwise, I
>> > > > > > think we're getting the default behavior where we generate
>> unique,
>> > > > > > per-process IDs, i.e. via this logic
>> > > > > >
>> > > > > >
>> > > > >
>> > > >
>> > >
>> >
>> https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java#L662-L664
>> > 

Re: Troubleshooting custom implementation for a org.apache.kafka.common.metrics.MetricsReporter

[Guozhang Wang]

Hello Artem,

Your observation is correct: for Streams as well as many other clients,
MetricsReporter#metricChange() are primarily used for registering new
metrics, this is because at the construction time of the client (hence when
MetricsReporter#init() is called) those finer-grained metrics, like
per-task / process, or per-destination-broker (for producer / consumer
clients, e.g.) are not known yet so they have to be created later during
runtime.

As for your second question: I'm not sure what do you mean by `metric
labels`, and how they are modified. A `MetricName` contains a name /
group-name, and a tags map, all of them should be fixed when being
registered.


Guozhang

On Fri, Feb 8, 2019 at 3:22 PM Artem Oboturov 
wrote:

> Hi
>
> I was checking out how to export metrics from a Kafka Steams App directly
> to the Prometheus Registry without using JMX, just by implementing a custom
> MetricsReporter.
>
> After some investigation, it became clear that a metric with the same name
> could be used by multiple entities, e.g. with different client-id. So it
> would be possible to differentiate them by that id.
>
> The MetricsReporter is configured for a Kafka Streams application.
>
> What I felt strange was that the MetricsReporter#init() is always called
> with an empty list, so it is not possible to define Prometheus metrics
> properly in advance and then keep them there to be updated with new values.
> Hence the MetricsReporter#metricChange() is used to both make changes to
> metered values and to define new metrics if they were not yet set up. Here
> comes a second problem: labels are variable, i.e. they are modified after
> they were first initialized, so that trying to set values, I often have the
> constraint violation in Prometheus, because it requires the labels set to
> be fixed.
>
> *ENV:*
> Kafka Streams (Scala): 2.1.0-cp1
> Kafka: Confluent.cloud, GCP
> Java: 8
>
> java.lang.IllegalArgumentException: Incorrect number of labels.
> at io.prometheus.client.SimpleCollector.labels(SimpleCollector.java:64)
> ~[simpleclient-0.6.0.jar:?]
> at
>
> privateimpl.KafkaStreamsPrometheusMetricsReporter.metricChange(KafkaStreamsPrometheusMetricsReporter.scala:82)
> ~[classes/:?]
> at org.apache.kafka.common.metrics.Metrics.registerMetric(Metrics.java:563)
> [kafka-clients-2.1.0-cp1.jar:?]
> at org.apache.kafka.common.metrics.Sensor.add(Sensor.java:278)
> [kafka-clients-2.1.0-cp1.jar:?]
> at org.apache.kafka.common.metrics.Sensor.add(Sensor.java:254)
> [kafka-clients-2.1.0-cp1.jar:?]
> at
>
> org.apache.kafka.clients.consumer.internals.Fetcher$FetchManagerMetrics.recordPartitionLead(Fetcher.java:1489)
> [kafka-clients-2.1.0-cp1.jar:?]
> at
>
> org.apache.kafka.clients.consumer.internals.Fetcher$FetchManagerMetrics.access$1600(Fetcher.java:1392)
> [kafka-clients-2.1.0-cp1.jar:?]
> at
>
> org.apache.kafka.clients.consumer.internals.Fetcher.fetchRecords(Fetcher.java:557)
> [kafka-clients-2.1.0-cp1.jar:?]
> at
>
> org.apache.kafka.clients.consumer.internals.Fetcher.fetchedRecords(Fetcher.java:505)
> [kafka-clients-2.1.0-cp1.jar:?]
> at
>
> org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1256)
> [kafka-clients-2.1.0-cp1.jar:?]
> at
>
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1188)
> [kafka-clients-2.1.0-cp1.jar:?]
> at
>
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1164)
> [kafka-clients-2.1.0-cp1.jar:?]
> at
>
> org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:913)
> [kafka-streams-2.1.0-cp1.jar:?]
> at
>
> org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:822)
> [kafka-streams-2.1.0-cp1.jar:?]
> at
>
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:777)
> [kafka-streams-2.1.0-cp1.jar:?]
> at
>
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:747)
> [kafka-streams-2.1.0-cp1.jar:?]
>
> Test implementation of the MetricsReporter:
>
> package privateimpl
>
> import java.util
> import java.util.concurrent.ConcurrentHashMap
>
> import com.typesafe.scalalogging.LazyLogging
> import privateimpl.KafkaStreamsPrometheusMetricsReporter.MKey
> import io.prometheus.client.{Collector, CollectorRegistry, Gauge}
> import org.apache.kafka.common.MetricName
> import org.apache.kafka.common.metrics.KafkaMetric
>
> import scala.util.{Failure, Success, Try}
>
> object KafkaStreamsPrometheusMetricsReporter {
> type MKey = (String, String)
> //  type MKey = MetricName
>
>   private def toKey(metric: KafkaMetric): MKey = {
> val name = metric.metricName()
> (name.name(), name.group())
> //name
>   }
>
>   private def toPrometheusMetric(metric: KafkaMetric): Gauge = {
> val name = metric.metricName()
> val labels = name.tags().keySet().toArray(Array.empty[String]).map {
>   Collector.sanitizeMetricName
> }
>
> Gauge
>   

Re: [VOTE] KIP-428: Add in-memory window store

[Sophie Blee-Goldman]

With three +1 (binding) votes this KIP is now accepted.

Thanks for the votes everyone.

-Sophie

On Wed, Feb 20, 2019 at 7:45 AM Bill Bejeck  wrote:

> Hi Sophie,
>
> Thanks for the KIP! My apologies for the delay in voting.
>
> +1(binding)
>
> -Bill
>
> On Thu, Feb 14, 2019 at 8:59 PM Guozhang Wang  wrote:
>
> > +1 (binding).
> >
> > On Thu, Feb 14, 2019 at 4:07 PM Matthias J. Sax 
> > wrote:
> >
> > > +1 (binding)
> > >
> > >
> > > -Matthias
> > >
> > > On 2/14/19 3:36 PM, Sophie Blee-Goldman wrote:
> > > > Hi all,
> > > >
> > > > I would like to call for a vote on KIP-428 regarding adding an
> > in-memory
> > > > version of the window store.
> > > >
> > > > The KIP can be found here:
> > > >
> > >
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-428%3A+Add+in-memory+window+store
> > > >
> > > > Cheers,
> > > > Sophie
> > > >
> > >
> > >
> >
> > --
> > -- Guozhang
> >
>

[jira] [Created] (KAFKA-7961) Handle subscription changes with a rebalance in progress

[Jason Gustafson (JIRA)]

Jason Gustafson created KAFKA-7961:
--

 Summary: Handle subscription changes with a rebalance in progress
 Key: KAFKA-7961
 URL: https://issues.apache.org/jira/browse/KAFKA-7961
 Project: Kafka
  Issue Type: Bug
Reporter: Jason Gustafson
Assignee: Jason Gustafson


Due to wakeups or poll timeouts, it is possible to have a subscription changed 
while a rebalance is in progress. This can lead to an illegal state error such 
as the following if some of the assigned partitions no longer match the 
subscription:

{code}
java.lang.IllegalArgumentException: Assigned partition foo-0 for non-subscribed 
topic; subscription is [bar]
at 
org.apache.kafka.clients.consumer.internals.SubscriptionState.assignFromSubscribed(SubscriptionState.java:192)
at 
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:249)
at 
org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:410)
at 
org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:344)
at 
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:344)
at 
org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1226)
at 
org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1191)
at 
org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1176)
{code}

Rather than requiring the assignment received from a rebalance to match the 
subscription, we should just request a rebalance.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (KAFKA-7960) KIP-432: Additional Broker-Side Opt-In for Default, Unsecure SASL/OAUTHBEARER Implementation

[Ron Dagostino (JIRA)]

Ron Dagostino created KAFKA-7960:


 Summary: KIP-432: Additional Broker-Side Opt-In for Default, 
Unsecure SASL/OAUTHBEARER Implementation
 Key: KAFKA-7960
 URL: https://issues.apache.org/jira/browse/KAFKA-7960
 Project: Kafka
  Issue Type: Improvement
  Components: clients
Affects Versions: 2.1.1, 2.1.0, 2.0.1, 2.0.0, 2.2.0, 2.1.2
Reporter: Ron Dagostino


The default implementation of SASL/OAUTHBEARER, as per KIP-255 
(https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=75968876), is 
unsecured.  This is useful for development and testing purposes, and it 
provides a great out-of-the-box experience, but it must not be used in 
production because it allows the client to authenticate with any principal name 
it wishes.  To enable the default unsecured SASL/OAUTHBEARER implementation on 
the broker side simply requires the addition of OAUTHBEARER to the 
sasl.enabled.mechanisms configuration value (for example: 
sasl.enabled.mechanisms=GSSAPI,OAUTHBEARER instead of simply 
sasl.enabled.mechanisms=GSSAPI). To secure the implementation requires the 
explicit setting of the 
listener.name.{sasl_plaintext|sasl_ssl}.oauthbearer.sasl.{login,server}.callback.handler.class
 properties on the broker.  The question then arises: what if someone either 
accidentally or maliciously appended OAUTHBEARER to the sasl.enabled.mechanisms 
configuration value?  Doing so would enable the unsecured implementation on the 
broker, and clients could then authenticate with any principal name they 
desired.

KIP-432 
(https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=103091238) 
proposes to add an additional opt-in configuration property on the broker side 
for the default, unsecured SASL/OAUTHBEARER implementation such that simply 
adding OAUTHBEARER to the sasl.enabled.mechanisms configuration value would be 
insufficient to enable the feature.  This additional opt-in broker 
configuration property would have to be explicitly set to true before the 
default unsecured implementation would successfully authenticate users, and the 
name of this configuration property would explicitly indicate that the feature 
is not secure and must not be used in production.  Adding this explicit opt-in 
is a breaking change; existing uses of the unsecured implementation would have 
to update their configuration to include this explicit opt-in property before 
their cluster would accept unsecure tokens again.  Note that this would only 
result in a breaking change in production if the unsecured feature is either 
accidentally or maliciously enabled there; it is assumed that 1) this will 
probably not happen to anyone; and 2) if it does happen to someone it almost 
certainly would not impact sanctioned clients but would instead impact 
malicious clients only (if there were any).



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[DISCUSS] KIP-432: Additional Broker-Side Opt-In for Default, Unsecure SASL/OAUTHBEARER Implementation

[Ron Dagostino]

Hi everyone. I created KIP-432: Additional Broker-Side Opt-In for Default,
Unsecure SASL/OAUTHBEARER Implementation

 (https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=103091238).
The motivation for this KIPis as follows:

The default implementation of SASL/OAUTHBEARER, as per KIP-255
,
is unsecured.  This is useful for development and testing purposes, and it
provides a great out-of-the-box experience, but it must not be used in
production because it allows the client to authenticate with any principal
name it wishes.  To enable the default unsecured SASL/OAUTHBEARER
implementation on the broker side simply requires the addition of
OAUTHBEARER to the sasl.enabled.mechanisms configuration value (for example:
 sasl.enabled.mechanisms=GSSAPI,OAUTHBEARER instead of simply
sasl.enabled.mechanisms=GSSAPI). To secure the implementation requires the
explicit setting of the
listener.name.{sasl_plaintext|sasl_ssl}.oauthbearer.sasl.{login,server}.callback.handler.class
 properties on the broker.  The question then arises: what if someone
either accidentally or maliciously appended OAUTHBEARER to the
sasl.enabled.mechanisms configuration value?  Doing so would enable the
unsecured implementation on the broker, and clients could then authenticate
with any principal name they desired.

This KIP proposes to add an additional opt-in configuration property on the
broker side for the default, unsecured SASL/OAUTHBEARER implementation such
that simply adding OAUTHBEARER to the sasl.enabled.mechanisms configuration
value would be insufficient to enable the feature.  This additional opt-in
broker configuration property would have to be explicitly set to true
before the default unsecured implementation would successfully authenticate
users, and the name of this configuration property would explicitly
indicate that the feature is not secure and must not be used in
production.  Adding this explicit opt-in is a breaking change; existing
uses of the unsecured implementation would have to update their
configuration to include this explicit opt-in property before their cluster
would accept unsecure tokens again.  Note that this would only result in a
breaking change in production if the unsecured feature is either
accidentally or maliciously enabled there; it is assumed that 1) this will
probably not happen to anyone; and 2) if it does happen to someone it
almost certainly would not impact sanctioned clients but would instead
impact malicious clients only (if there were any).


Ron

[jira] [Created] (KAFKA-7959) Clear/delete epoch cache if old message format is in use

[Jason Gustafson (JIRA)]

Jason Gustafson created KAFKA-7959:
--

 Summary: Clear/delete epoch cache if old message format is in use
 Key: KAFKA-7959
 URL: https://issues.apache.org/jira/browse/KAFKA-7959
 Project: Kafka
  Issue Type: Bug
Reporter: Jason Gustafson


Because of KAFKA-7897, it is possible to have a sparse epoch cache when using 
the old message format. The fix for that issue addresses the problem of 
improper use of that cache while the message format remains on an older 
version. However, it leaves the possibility of misuse during a message format 
upgrade, which can cause unexpected truncation and re-replication. To fix the 
problem, we should delete or at least clear the cache whenever the old message 
format is used.

Note that this problem was fixed unintentionally in 2.1 with the patch for 
KAFKA-7897. This issue applies specifically to the 2.0 branch.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Re: [VOTE] KIP-428: Add in-memory window store

[Bill Bejeck]

Hi Sophie,

Thanks for the KIP! My apologies for the delay in voting.

+1(binding)

-Bill

On Thu, Feb 14, 2019 at 8:59 PM Guozhang Wang  wrote:

> +1 (binding).
>
> On Thu, Feb 14, 2019 at 4:07 PM Matthias J. Sax 
> wrote:
>
> > +1 (binding)
> >
> >
> > -Matthias
> >
> > On 2/14/19 3:36 PM, Sophie Blee-Goldman wrote:
> > > Hi all,
> > >
> > > I would like to call for a vote on KIP-428 regarding adding an
> in-memory
> > > version of the window store.
> > >
> > > The KIP can be found here:
> > >
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-428%3A+Add+in-memory+window+store
> > >
> > > Cheers,
> > > Sophie
> > >
> >
> >
>
> --
> -- Guozhang
>

Re: [DISCUSS] KIP-430 - Return Authorized Operations in Describe Responses

[Rajini Sivaram]

If there are no other concerns or suggestions, I will start vote on this
KIP later today.

Thanks,

Rajini

On Mon, Feb 18, 2019 at 10:09 AM Rajini Sivaram 
wrote:

> Hi Magnus,
>
> Have your concerns been addressed in the KIP?
>
> Thanks,
>
> Rajini
>
> On Wed, Feb 13, 2019 at 3:33 PM Satish Duggana 
> wrote:
>
>> Hi Rajini,
>> That makes sense, thanks for the clarification.
>>
>> Satish.
>>
>> On Wed, Feb 13, 2019 at 7:30 PM Rajini Sivaram 
>> wrote:
>> >
>> > Thanks for the reviews!
>> >
>> > Hi Satish,
>> >
>> > The authorised operations returned will use the same values as the
>> > operations returned by the existing DescribeAclsResponse. AdminClient
>> will
>> > return these using the existing enum AclOperation.
>> >
>> > Hi Magnus,
>> >
>> > The MetadataResponse contains these two lines:
>> >
>> >- Metadata Response => throttle_time_ms [brokers] cluster_id
>> >controller_id [topic_metadata] [authorized_operations] <== ADDED
>> >authorized_operations
>> >- topic_metadata => error_code topic is_internal [partition_metadata]
>> >[authorized_operations]  <== ADDED authorized_operations
>> >
>> > The first is for the cluster's authorized operations and the second for
>> > each topic. Did I misunderstand your question? The full set of
>> operations
>> > for each resource type is included in the subsection `AdminClient API
>> > Changes`.
>> >
>> > Under `Rejected Alternatives` I have included addition of a separate
>> > request to get this information rather than extend an existing one. The
>> > rationale for including all the information in one request is to enable
>> > clients to get all relevant metadata using a single API rather than
>> have to
>> > send multiple requests, get responses and combine the two while
>> resource or
>> > ACLs may have changed in between. It seems neater to use a single API
>> since
>> > a user getting authorized operations is almost definitely going to do a
>> > Describe first and access control for both of these is controlled using
>> > Describe access. If we add new resource types with a corresponding
>> > Describe, we would simply need to add `authorized_operations` for their
>> > Describe.
>> >
>> > Hi Manikumar,
>> >
>> > Added IdempotentWrite for Cluster, thanks for pointing that out! I was
>> > thinking that if authorizer is not configured, we could return all
>> > supported operations since the user can perform all operations. Added a
>> > note to the KIP.
>> >
>> > Regards,
>> >
>> > Rajini
>> >
>> >
>> >
>> > On Wed, Feb 13, 2019 at 11:07 AM Manikumar 
>> > wrote:
>> >
>> > > Hi,
>> > >
>> > > Thanks for the KIP.
>> > >
>> > > 1. Can't we include IdempotentWrite/ClusterResource Operations for
>> Cluster
>> > > resource.
>> > > 2. What will be the API behaviour when the authorizer is not
>> configured?. I
>> > > assume we return empty list.
>> > >
>> > > Thanks,
>> > > Manikumar
>> > >
>> > > On Wed, Feb 13, 2019 at 12:33 AM Rajini Sivaram <
>> rajinisiva...@gmail.com>
>> > > wrote:
>> > >
>> > > > Hi all,
>> > > >
>> > > > I have created a KIP to optionally request authorised operations on
>> > > > resources when describing resources:
>> > > >
>> > > >
>> > > >
>> > >
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-430+-+Return+Authorized+Operations+in+Describe+Responses
>> > > >
>> > > > This includes only information that users with Describe access can
>> obtain
>> > > > using other means and hence is consistent with our security model.
>> It is
>> > > > intended to made it easier for clients to obtain this information.
>> > > >
>> > > > Feedback and suggestions welcome.
>> > > >
>> > > > Thank you,
>> > > >
>> > > > Rajini
>> > > >
>> > >
>>
>