[jira] [Commented] (KAFKA-3693) Race condition between highwatermark-checkpoint thread and handleLeaderAndIsrRequest at broker start-up

[Maysam Yabandeh (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-3693?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15315112#comment-15315112
 ] 

Maysam Yabandeh commented on KAFKA-3693:


As mentioned above:
bq. We managed to reproduce the problem by restarting all the nodes of a test 
cluster of size 10 at the same time
By restart I meant sending the interrupt signal to the brokers. So broker 16 
was shutting down at 06:17:02,012 (not down yet). I guess your expectation is 
that the controller have a mechanism to timeout on broker 16 and remove it from 
its broker lists? 

> Race condition between highwatermark-checkpoint thread and 
> handleLeaderAndIsrRequest at broker start-up
> ---
>
> Key: KAFKA-3693
> URL: https://issues.apache.org/jira/browse/KAFKA-3693
> Project: Kafka
>  Issue Type: Bug
>  Components: core
>Affects Versions: 0.9.0.1
>Reporter: Maysam Yabandeh
>
> Upon broker start-up, a race between highwatermark-checkpoint thread to write 
> replication-offset-checkpoint file and handleLeaderAndIsrRequest thread 
> reading from it causes the highwatermark for some partitions to be reset to 
> 0. In the good case, this results the replica to truncate its entire log to 0 
> and hence initiates fetching of terabytes of data from the lead broker, which 
> sometimes leads to hours of downtime. We observed the bad cases that the 
> reset offset can propagate to recovery-point-offset-checkpoint file, making a 
> lead broker to truncate the file. This seems to have the potential to lead to 
> data loss if the truncation happens at both follower and leader brokers.
> This is the particular faulty scenario manifested in our tests:
> # The broker restarts and receive LeaderAndIsr from the controller
> # LeaderAndIsr message however does not contain all the partitions (probably 
> because other brokers were churning at the same time)
> # becomeLeaderOrFollower calls getOrCreatePartition and updates the 
> allPartitions with the partitions included in the LeaderAndIsr message {code}
>   def getOrCreatePartition(topic: String, partitionId: Int): Partition = {
> var partition = allPartitions.get((topic, partitionId))
> if (partition == null) {
>   allPartitions.putIfNotExists((topic, partitionId), new Partition(topic, 
> partitionId, time, this))
> {code}
> # replication-offset-checkpoint jumps in taking a snapshot of (the partial) 
> allReplicas' high watermark into replication-offset-checkpoint file {code}  
> def checkpointHighWatermarks() {
> val replicas = 
> allPartitions.values.map(_.getReplica(config.brokerId)).collect{case 
> Some(replica) => replica}{code} hence rewriting the previous highwatermarks.
> # Later becomeLeaderOrFollower calls makeLeaders and makeFollowers which read 
> the (now partial) file through Partition::getOrCreateReplica {code}
>   val checkpoint = 
> replicaManager.highWatermarkCheckpoints(log.dir.getParentFile.getAbsolutePath)
>   val offsetMap = checkpoint.read
>   if (!offsetMap.contains(TopicAndPartition(topic, partitionId)))
> info("No checkpointed highwatermark is found for partition 
> [%s,%d]".format(topic, partitionId))
> {code}
> We are not entirely sure whether the initial LeaderAndIsr message including a 
> subset of partitions is critical in making this race condition manifest or 
> not. But it is an important detail since it clarifies that a solution based 
> on not letting the highwatermark-checkpoint thread jumping in the middle of 
> processing a LeaderAndIsr message would not suffice.
> The solution we are thinking of is to force initializing allPartitions by the 
> partitions listed in the replication-offset-checkpoint (and perhaps 
> recovery-point-offset-checkpoint file too) when a server starts.
> Thoughts?



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-3693) Race condition between highwatermark-checkpoint thread and handleLeaderAndIsrRequest at broker start-up

[Maysam Yabandeh (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-3693?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15315055#comment-15315055
 ] 

Maysam Yabandeh commented on KAFKA-3693:


Thanks [~junrao]. My understanding is that the LeaderAndIsr message with 
partial list has been failed to be delivered to broker 16 and it was finally 
delivered when broker 16 was back online again.

Let me paste more of relevant logs perhaps it would make the buggy scenario 
more clear.

At 06:17:02,012 the controller attempts to send the LeaderAndIsr message to 
broker 16

{code}
2016-05-10 06:17:02,012 TRACE change.logger: Controller 17 epoch 269 sending 
become-leader LeaderAndIsr request 
(Leader:16,ISR:16,LeaderEpoch:79,ControllerEpoch:269) to broker 16 for 
partition [topic.xyz,134]
{code}

This attempt however fails 
{code}
2016-05-10 06:17:02,076 WARN controller.RequestSendThread: 
[Controller-17-to-broker-16-send-thread], Controller 17's connection to broker 
Node(16, node16.com, 9876) was unsuccessful
{code}
since broker 16 has been shutting down since 05:40:46,845 until 06:17:01,701
{code}
2016-05-10 05:40:46,845 INFO server.ReplicaFetcherThread: 
[ReplicaFetcherThread-0-15], Shutting down
...
2016-05-10 06:17:01,701 
{code}

The controller repeats the attempt until it finally connects to broker 16 at 
06:17:33,344
{code}2016-05-10 06:17:33,344 INFO controller.RequestSendThread: 
[Controller-17-to-broker-16-send-thread], Controller 17 connected to Node(16, 
node16.com, 9876) for sending state change requests
{code}
when the broker 16 is restarted. 

Here is the first 5 state changes the broker 16 performs right after the 
restart:
{code}
2016-05-10 06:17:33,410 TRACE change.logger: Broker 16 cached leader info 
(LeaderAndIsrInfo:(Leader:21,ISR:21,LeaderEpoch:58,ControllerEpoch:269),ReplicationFactor:2),AllReplicas:20,21)
 for partition [topic.qwe,3] in response to UpdateMetadata request sent by 
controller 17 epoch 269 with correlation id 2398
2016-05-10 06:17:33,438 TRACE change.logger: Broker 16 cached leader info 
(LeaderAndIsrInfo:(Leader:14,ISR:14,LeaderEpoch:110,ControllerEpoch:269),ReplicationFactor:2),AllReplicas:20,14)
 for partition [topic.asd,88] in response to UpdateMetadata request sent by 
controller 17 epoch 269 with correlation id 2399
2016-05-10 06:17:33,440 TRACE change.logger: Broker 16 cached leader info 
(LeaderAndIsrInfo:(Leader:15,ISR:15,LeaderEpoch:18,ControllerEpoch:269),ReplicationFactor:2),AllReplicas:20,15)
 for partition [topic.zxc,8] in response to UpdateMetadata request sent by 
controller 17 epoch 269 with correlation id 2400
2016-05-10 06:17:33,442 TRACE change.logger: Broker 16 cached leader info 
(LeaderAndIsrInfo:(Leader:21,ISR:21,LeaderEpoch:61,ControllerEpoch:269),ReplicationFactor:2),AllReplicas:21,20)
 for partition [topic.iop,4] in response to UpdateMetadata request sent by 
controller 17 epoch 269 with correlation id 2401
2016-05-10 06:17:33,447 TRACE change.logger: Broker 16 received LeaderAndIsr 
request 
(LeaderAndIsrInfo:(Leader:16,ISR:16,LeaderEpoch:79,ControllerEpoch:269),ReplicationFactor:2),AllReplicas:16,20)
 correlation id 2402 from controller 17 epoch 269 for partition [topic.xyz,134]
{code}
The last of which is the LeaderAndIsr with incomplete list of partitions that 
we were talking about. I do not see any other track in the controller log 
indicating of any similar message sent to broker 16, which tells me the 
LeaderAndIsr message received at 06:17:33,447 must be the same one that was 
formed at 06:17:02,012 and has been attempted later when broker 16 was back 
online.

Does the above make sense?

> Race condition between highwatermark-checkpoint thread and 
> handleLeaderAndIsrRequest at broker start-up
> ---
>
> Key: KAFKA-3693
> URL: https://issues.apache.org/jira/browse/KAFKA-3693
> Project: Kafka
>  Issue Type: Bug
>  Components: core
>Affects Versions: 0.9.0.1
>Reporter: Maysam Yabandeh
>
> Upon broker start-up, a race between highwatermark-checkpoint thread to write 
> replication-offset-checkpoint file and handleLeaderAndIsrRequest thread 
> reading from it causes the highwatermark for some partitions to be reset to 
> 0. In the good case, this results the replica to truncate its entire log to 0 
> and hence initiates fetching of terabytes of data from the lead broker, which 
> sometimes leads to hours of downtime. We observed the bad cases that the 
> reset offset can propagate to recovery-point-offset-checkpoint file, making a 
> lead broker to truncate the file. This seems to have the potential to lead to 
> data loss if the truncation happens at both follower and leader brokers.
> This is the particular faulty scenario manifested in our tests:
> # The broker restarts and receive LeaderAndIsr from the controller
> # 

Re: [DISCUSS] KIP-63: Unify store and downstream caching in streams

[Jay Kreps]

Hey Eno,

Should the config be the global memory use rather than the per-processor?
That is, let’s say I know I have fixed a 1GB heap because that is what I
set for Java, and want to use 100MB for caching, it seems like right now
I’d have to do some math that depends on my knowing a bit about how caching
works to figure out how to set that parameter so I don't run out of memory.
Does it also depend on the number of partitions assigned (and hence the
number of task), if so that makes it even harder to set since each time
rebalancing happens that changes so it is then pretty hard to set safely.

You could theoretically argue for either bottom up (you know how much cache
you need per processor as you have it and you want to get exactly that) or
top down (you know how much memory you have to spare but can't be bothered
to work out what that amounts to per-processor). I think our experience has
been that 99% of people never change the default and if it runs out of
memory they really struggle to fix it and kind of blame us, so I think top
down and a global config might be better. :-)

Example: https://issues.apache.org/jira/browse/KAFKA-3775

-Jay

On Fri, Jun 3, 2016 at 2:39 PM, Eno Thereska  wrote:

> Hi Gwen,
>
> Yes. As an example, if cache.max.bytes.buffering set to X, and if users
> have A aggregation operators and T KTable.to() operators, then X*(A + T)
> total bytes will be allocated for caching.
>
> Eno
>
> > On 3 Jun 2016, at 21:37, Gwen Shapira  wrote:
> >
> > Just to clarify: "cache.max.bytes.buffering" is per processor?
> >
> >
> > On Thu, Jun 2, 2016 at 11:30 AM, Eno Thereska 
> wrote:
> >> Hi there,
> >>
> >> I have created KIP-63: Unify store and downstream caching in streams
> >>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-63%3A+Unify+store+and+downstream+caching+in+streams
> <
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-63:+Unify+store+and+downstream+caching+in+streams
> >
> >>
> >>
> >> Feedback is appreciated.
> >>
> >> Thank you
> >> Eno
>
>

Re: [DISCUSS] KIP-63: Unify store and downstream caching in streams

[Roger Hoover]

Hi Eno,

Does this mean that Kafka Streams will disable the RocksDB write buffer?

Is it currently safe to recover a Kafka Streams application after SIGKILL
on the same machine?  If not, will this make it safe to do so?

If RocksDB is not flushed before offsets are commited in Kafka and is
killed with SIGKILL, will the data in the write buffer be lost (since Kafka
Streams disables the transaction log)?  That data will be present in the
Kafka changelog but will it get applied to the recovered RocksDB database
on restart?

Thanks,

Roger

On Fri, Jun 3, 2016 at 2:39 PM, Eno Thereska  wrote:

> Hi Gwen,
>
> Yes. As an example, if cache.max.bytes.buffering set to X, and if users
> have A aggregation operators and T KTable.to() operators, then X*(A + T)
> total bytes will be allocated for caching.
>
> Eno
>
> > On 3 Jun 2016, at 21:37, Gwen Shapira  wrote:
> >
> > Just to clarify: "cache.max.bytes.buffering" is per processor?
> >
> >
> > On Thu, Jun 2, 2016 at 11:30 AM, Eno Thereska 
> wrote:
> >> Hi there,
> >>
> >> I have created KIP-63: Unify store and downstream caching in streams
> >>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-63%3A+Unify+store+and+downstream+caching+in+streams
> <
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-63:+Unify+store+and+downstream+caching+in+streams
> >
> >>
> >>
> >> Feedback is appreciated.
> >>
> >> Thank you
> >> Eno
>
>

Re: [DISCUSS] KIP-63: Unify store and downstream caching in streams

[Eno Thereska]

Hi Gwen,

Yes. As an example, if cache.max.bytes.buffering set to X, and if users have A 
aggregation operators and T KTable.to() operators, then X*(A + T) total bytes 
will be allocated for caching.

Eno

> On 3 Jun 2016, at 21:37, Gwen Shapira  wrote:
> 
> Just to clarify: "cache.max.bytes.buffering" is per processor?
> 
> 
> On Thu, Jun 2, 2016 at 11:30 AM, Eno Thereska  wrote:
>> Hi there,
>> 
>> I have created KIP-63: Unify store and downstream caching in streams
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-63%3A+Unify+store+and+downstream+caching+in+streams
>>  
>> 
>> 
>> 
>> Feedback is appreciated.
>> 
>> Thank you
>> Eno

Build failed in Jenkins: kafka-0.10.0-jdk7 #115

[Apache Jenkins Server]

See 

Changes:

[wangguoz] KAFKA-3784: TimeWindows#windowsFor calculation is incorrect

[ismael] KAFKA-3728; EndToEndAuthorizationTest offsets_topic misconfigured

--
Started by an SCM change
[EnvInject] - Loading node environment variables.
Building remotely on H11 (docker Ubuntu ubuntu yahoo-not-h2) in workspace 

 > git rev-parse --is-inside-work-tree # timeout=10
Fetching changes from the remote Git repository
 > git config remote.origin.url 
 > https://git-wip-us.apache.org/repos/asf/kafka.git # timeout=10
Fetching upstream changes from https://git-wip-us.apache.org/repos/asf/kafka.git
 > git --version # timeout=10
 > git -c core.askpass=true fetch --tags --progress 
 > https://git-wip-us.apache.org/repos/asf/kafka.git 
 > +refs/heads/*:refs/remotes/origin/*
 > git rev-parse refs/remotes/origin/0.10.0^{commit} # timeout=10
 > git rev-parse refs/remotes/origin/origin/0.10.0^{commit} # timeout=10
Checking out Revision 717c7bbc39b56a17a413b3bd02c776d9b79e9b3e 
(refs/remotes/origin/0.10.0)
 > git config core.sparsecheckout # timeout=10
 > git checkout -f 717c7bbc39b56a17a413b3bd02c776d9b79e9b3e
 > git rev-list 419e6517c75822d94974628151d399ae93e4a9f0 # timeout=10
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK_1_7U51_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk-1.7u51
[kafka-0.10.0-jdk7] $ /bin/bash -xe /tmp/hudson8651382302142558795.sh
+ 
/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2/bin/gradle
To honour the JVM settings for this build a new JVM will be forked. Please 
consider using the daemon: 
http://gradle.org/docs/2.4-rc-2/userguide/gradle_daemon.html.
Building project 'core' with Scala version 2.10.6
:downloadWrapper

BUILD SUCCESSFUL

Total time: 26.272 secs
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK_1_7U51_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk-1.7u51
[kafka-0.10.0-jdk7] $ /bin/bash -xe /tmp/hudson172501671292958353.sh
+ export GRADLE_OPTS=-Xmx1024m
+ GRADLE_OPTS=-Xmx1024m
+ ./gradlew -Dorg.gradle.project.maxParallelForks=1 clean jarAll testAll
To honour the JVM settings for this build a new JVM will be forked. Please 
consider using the daemon: 
https://docs.gradle.org/2.13/userguide/gradle_daemon.html.
Building project 'core' with Scala version 2.10.6
Build file ': 
line 233
useAnt has been deprecated and is scheduled to be removed in Gradle 3.0. The 
Ant-Based Scala compiler is deprecated, please see 
https://docs.gradle.org/current/userguide/scala_plugin.html.
:clean UP-TO-DATE
:clients:clean
:connect:clean UP-TO-DATE
:core:clean
:examples:clean
:log4j-appender:clean
:streams:clean
:tools:clean
:connect:api:clean
:connect:file:clean
:connect:json:clean
:connect:runtime:clean
:streams:examples:clean
:jar_core_2_10
Building project 'core' with Scala version 2.10.6
:kafka-0.10.0-jdk7:clients:compileJava
:jar_core_2_10 FAILED

FAILURE: Build failed with an exception.

* What went wrong:
Failed to capture snapshot of input files for task 'compileJava' during 
up-to-date check.
> Could not add entry 
> '/home/jenkins/.gradle/caches/modules-2/files-2.1/net.jpountz.lz4/lz4/1.3.0/c708bb2590c0652a642236ef45d9f99ff842a2ce/lz4-1.3.0.jar'
>  to cache fileHashes.bin 
> (

* Try:
Run with --stacktrace option to get the stack trace. Run with --info or --debug 
option to get more log output.

BUILD FAILED

Total time: 23.736 secs
Build step 'Execute shell' marked build as failure
Recording test results
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK_1_7U51_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk-1.7u51
ERROR: Step ?Publish JUnit test result report? failed: No test report files 
were found. Configuration error?
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK_1_7U51_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk-1.7u51

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

[Apache Jenkins Server]

See 

Changes:

[wangguoz] KAFKA-3784: TimeWindows#windowsFor calculation is incorrect

[ismael] KAFKA-3728; EndToEndAuthorizationTest offsets_topic misconfigured

--
Started by an SCM change
[EnvInject] - Loading node environment variables.
Building remotely on ubuntu-us1 (Ubuntu ubuntu ubuntu-us golang-ppa) in 
workspace 
 > git rev-parse --is-inside-work-tree # timeout=10
Fetching changes from the remote Git repository
 > git config remote.origin.url 
 > https://git-wip-us.apache.org/repos/asf/kafka.git # timeout=10
Fetching upstream changes from https://git-wip-us.apache.org/repos/asf/kafka.git
 > git --version # timeout=10
 > git -c core.askpass=true fetch --tags --progress 
 > https://git-wip-us.apache.org/repos/asf/kafka.git 
 > +refs/heads/*:refs/remotes/origin/*
 > git rev-parse refs/remotes/origin/trunk^{commit} # timeout=10
 > git rev-parse refs/remotes/origin/origin/trunk^{commit} # timeout=10
Checking out Revision 49ddc897b8feda9c4786d5bcd03814b91ede7124 
(refs/remotes/origin/trunk)
 > git config core.sparsecheckout # timeout=10
 > git checkout -f 49ddc897b8feda9c4786d5bcd03814b91ede7124
 > git rev-list a27030693bd89b16a436ba873192ebc98c4c7910 # timeout=10
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK1_8_0_66_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk1.8.0_66
[kafka-trunk-jdk8] $ /bin/bash -xe /tmp/hudson7318532659157633030.sh
+ 
/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2/bin/gradle
To honour the JVM settings for this build a new JVM will be forked. Please 
consider using the daemon: 
http://gradle.org/docs/2.4-rc-2/userguide/gradle_daemon.html.
Building project 'core' with Scala version 2.10.6
:downloadWrapper

BUILD SUCCESSFUL

Total time: 19.512 secs
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK1_8_0_66_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk1.8.0_66
[kafka-trunk-jdk8] $ /bin/bash -xe /tmp/hudson9212758006965465175.sh
+ export GRADLE_OPTS=-Xmx1024m
+ GRADLE_OPTS=-Xmx1024m
+ ./gradlew -Dorg.gradle.project.maxParallelForks=1 
-Dorg.gradle.project.testLoggingEvents=started,passed,skipped,failed clean 
jarAll testAll
To honour the JVM settings for this build a new JVM will be forked. Please 
consider using the daemon: 
https://docs.gradle.org/2.13/userguide/gradle_daemon.html.
Building project 'core' with Scala version 2.10.6
Build file ': 
line 239
useAnt has been deprecated and is scheduled to be removed in Gradle 3.0. The 
Ant-Based Scala compiler is deprecated, please see 
https://docs.gradle.org/current/userguide/scala_plugin.html.
:clean UP-TO-DATE
:clients:clean UP-TO-DATE
:connect:clean UP-TO-DATE
:core:clean
:examples:clean UP-TO-DATE
:log4j-appender:clean UP-TO-DATE
:streams:clean UP-TO-DATE
:tools:clean UP-TO-DATE
:connect:api:clean UP-TO-DATE
:connect:file:clean UP-TO-DATE
:connect:json:clean UP-TO-DATE
:connect:runtime:clean UP-TO-DATE
:streams:examples:clean UP-TO-DATE
:jar_core_2_10
Building project 'core' with Scala version 2.10.6
:kafka-trunk-jdk8:clients:compileJava
:jar_core_2_10 FAILED

FAILURE: Build failed with an exception.

* What went wrong:
Failed to capture snapshot of input files for task 'compileJava' during 
up-to-date check.
> Could not add entry 
> '/home/jenkins/.gradle/caches/modules-2/files-2.1/net.jpountz.lz4/lz4/1.3.0/c708bb2590c0652a642236ef45d9f99ff842a2ce/lz4-1.3.0.jar'
>  to cache fileHashes.bin 
> (

* Try:
Run with --stacktrace option to get the stack trace. Run with --info or --debug 
option to get more log output.

BUILD FAILED

Total time: 13.595 secs
Build step 'Execute shell' marked build as failure
Recording test results
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK1_8_0_66_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk1.8.0_66
ERROR: Step ‘Publish JUnit test result report’ failed: No test report files 
were found. Configuration error?
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK1_8_0_66_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk1.8.0_66

Kafka 0.10.0.0 broker configuration to support older consumer/producer clients...

[Kiran Pillarisetty]

We are in the process of upgrading the broker to Kafka 0.10.0.0.  The
following documentation is not too clear to us:
"By setting the message format version, one certifies that all
existing messages are on or below that message format version.
Otherwise consumers before 0.10.0.0 might break."
http://kafka.apache.org/0100/documentation.html#upgrade_10_performance_impact

What exactly does that mean? Do we have to explicitly set
"log.message.format.version" on the broker to 0.10.0.0, and if we
don't, older consumer clients break? Note that our consumer and
producer clients are still on older versions (0.8.2.X, 0.9.0.X) at
this point, and we need to continue supporting them for a while.

Thanks in advance!
Kiran

[jira] [Resolved] (KAFKA-3728) EndToEndAuthorizationTest offsets_topic misconfigured

[Ismael Juma (JIRA)]

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

Ismael Juma resolved KAFKA-3728.

   Resolution: Fixed
Fix Version/s: 0.10.0.1

Issue resolved by pull request 1425
[https://github.com/apache/kafka/pull/1425]

> EndToEndAuthorizationTest offsets_topic misconfigured
> -
>
> Key: KAFKA-3728
> URL: https://issues.apache.org/jira/browse/KAFKA-3728
> Project: Kafka
>  Issue Type: Bug
>Reporter: Edoardo Comar
>Assignee: Edoardo Comar
> Fix For: 0.10.0.1
>
>
> A consumer that is manually assigned a topic-partition is able to consume 
> messages that a consumer that subscribes to the topic can not.
> To reproduce : take the test 
> EndToEndAuthorizationTest.testProduceConsume 
> (eg the SaslSslEndToEndAuthorizationTest implementation)
>  
> it passes ( = messages are consumed) 
> if the consumer is assigned the single topic-partition
>   consumers.head.assign(List(tp).asJava)
> but fails 
> if the consumer subscribes to the topic - changing the line to :
>   consumers.head.subscribe(List(topic).asJava)
> The failure when subscribed shows this error about synchronization:
>  org.apache.kafka.common.KafkaException: Unexpected error from SyncGroup: 
> Messages are rejected since there are fewer in-sync replicas than required.
>   at 
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator$SyncGroupResponseHandler.handle(AbstractCoordinator.java:455)
> The test passes in both cases (subscribe and assign) with the setting
>   this.serverConfig.setProperty(KafkaConfig.MinInSyncReplicasProp, "1")



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Build failed in Jenkins: kafka-trunk-jdk7 #1338

[Apache Jenkins Server]

See 

Changes:

[wangguoz] KAFKA-3784: TimeWindows#windowsFor calculation is incorrect

[ismael] KAFKA-3728; EndToEndAuthorizationTest offsets_topic misconfigured

--
Started by an SCM change
[EnvInject] - Loading node environment variables.
Building remotely on ubuntu3 (Ubuntu ubuntu legacy-ubuntu) in workspace 

 > git rev-parse --is-inside-work-tree # timeout=10
Fetching changes from the remote Git repository
 > git config remote.origin.url 
 > https://git-wip-us.apache.org/repos/asf/kafka.git # timeout=10
Fetching upstream changes from https://git-wip-us.apache.org/repos/asf/kafka.git
 > git --version # timeout=10
 > git -c core.askpass=true fetch --tags --progress 
 > https://git-wip-us.apache.org/repos/asf/kafka.git 
 > +refs/heads/*:refs/remotes/origin/*
 > git rev-parse refs/remotes/origin/trunk^{commit} # timeout=10
 > git rev-parse refs/remotes/origin/origin/trunk^{commit} # timeout=10
Checking out Revision 49ddc897b8feda9c4786d5bcd03814b91ede7124 
(refs/remotes/origin/trunk)
 > git config core.sparsecheckout # timeout=10
 > git checkout -f 49ddc897b8feda9c4786d5bcd03814b91ede7124
 > git rev-list a27030693bd89b16a436ba873192ebc98c4c7910 # timeout=10
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK_1_7U51_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk-1.7u51
[kafka-trunk-jdk7] $ /bin/bash -xe /tmp/hudson2869788452859777959.sh
+ 
/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2/bin/gradle
To honour the JVM settings for this build a new JVM will be forked. Please 
consider using the daemon: 
http://gradle.org/docs/2.4-rc-2/userguide/gradle_daemon.html.
Building project 'core' with Scala version 2.10.6
:downloadWrapper

BUILD SUCCESSFUL

Total time: 19.978 secs
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK_1_7U51_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk-1.7u51
[kafka-trunk-jdk7] $ /bin/bash -xe /tmp/hudson4992425481380617329.sh
+ export GRADLE_OPTS=-Xmx1024m
+ GRADLE_OPTS=-Xmx1024m
+ ./gradlew -Dorg.gradle.project.maxParallelForks=1 
-Dorg.gradle.project.testLoggingEvents=started,passed,skipped,failed clean 
jarAll testAll
To honour the JVM settings for this build a new JVM will be forked. Please 
consider using the daemon: 
https://docs.gradle.org/2.13/userguide/gradle_daemon.html.
Building project 'core' with Scala version 2.10.6
Build file '/x1/jenkins/jenkins-slave/workspace/kafka-trunk-jdk7/build.gradle': 
line 239
useAnt has been deprecated and is scheduled to be removed in Gradle 3.0. The 
Ant-Based Scala compiler is deprecated, please see 
https://docs.gradle.org/current/userguide/scala_plugin.html.
:clean UP-TO-DATE
:clients:clean
:connect:clean UP-TO-DATE
:core:clean
:examples:clean
:log4j-appender:clean
:streams:clean
:tools:clean
:connect:api:clean
:connect:file:clean
:connect:json:clean
:connect:runtime:clean
:streams:examples:clean
:jar_core_2_10
Building project 'core' with Scala version 2.10.6
:kafka-trunk-jdk7:clients:compileJava
:jar_core_2_10 FAILED

FAILURE: Build failed with an exception.

* What went wrong:
Failed to capture snapshot of input files for task 'compileJava' during 
up-to-date check.
> Could not add entry 
> '/home/jenkins/.gradle/caches/modules-2/files-2.1/net.jpountz.lz4/lz4/1.3.0/c708bb2590c0652a642236ef45d9f99ff842a2ce/lz4-1.3.0.jar'
>  to cache fileHashes.bin 
> (/x1/jenkins/jenkins-slave/workspace/kafka-trunk-jdk7/.gradle/2.13/taskArtifacts/fileHashes.bin).

* Try:
Run with --stacktrace option to get the stack trace. Run with --info or --debug 
option to get more log output.

BUILD FAILED

Total time: 17.907 secs
Build step 'Execute shell' marked build as failure
Recording test results
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK_1_7U51_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk-1.7u51
ERROR: Step ?Publish JUnit test result report? failed: No test report files 
were found. Configuration error?
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK_1_7U51_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk-1.7u51

Re: [DISCUSS] KIP-63: Unify store and downstream caching in streams

[Gwen Shapira]

Just to clarify: "cache.max.bytes.buffering" is per processor?


On Thu, Jun 2, 2016 at 11:30 AM, Eno Thereska  wrote:
> Hi there,
>
> I have created KIP-63: Unify store and downstream caching in streams
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-63%3A+Unify+store+and+downstream+caching+in+streams
>  
> 
>
>
> Feedback is appreciated.
>
> Thank you
> Eno

[jira] [Commented] (KAFKA-3728) EndToEndAuthorizationTest offsets_topic misconfigured

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-3728?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15314748#comment-15314748
 ] 

ASF GitHub Bot commented on KAFKA-3728:
---

Github user asfgit closed the pull request at:

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


> EndToEndAuthorizationTest offsets_topic misconfigured
> -
>
> Key: KAFKA-3728
> URL: https://issues.apache.org/jira/browse/KAFKA-3728
> Project: Kafka
>  Issue Type: Bug
>Reporter: Edoardo Comar
>Assignee: Edoardo Comar
>
> A consumer that is manually assigned a topic-partition is able to consume 
> messages that a consumer that subscribes to the topic can not.
> To reproduce : take the test 
> EndToEndAuthorizationTest.testProduceConsume 
> (eg the SaslSslEndToEndAuthorizationTest implementation)
>  
> it passes ( = messages are consumed) 
> if the consumer is assigned the single topic-partition
>   consumers.head.assign(List(tp).asJava)
> but fails 
> if the consumer subscribes to the topic - changing the line to :
>   consumers.head.subscribe(List(topic).asJava)
> The failure when subscribed shows this error about synchronization:
>  org.apache.kafka.common.KafkaException: Unexpected error from SyncGroup: 
> Messages are rejected since there are fewer in-sync replicas than required.
>   at 
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator$SyncGroupResponseHandler.handle(AbstractCoordinator.java:455)
> The test passes in both cases (subscribe and assign) with the setting
>   this.serverConfig.setProperty(KafkaConfig.MinInSyncReplicasProp, "1")



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] kafka pull request #1425: KAFKA-3728 EndToEndAuthorizationTest offsets_topic...

[asfgit]

Github user asfgit closed the pull request at:

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


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Resolved] (KAFKA-3784) TimeWindows#windowsFor misidentifies some windows if TimeWindows#advanceBy is used

[Guozhang Wang (JIRA)]

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

Guozhang Wang resolved KAFKA-3784.
--
   Resolution: Fixed
Fix Version/s: 0.10.0.1

Issue resolved by pull request 1462
[https://github.com/apache/kafka/pull/1462]

> TimeWindows#windowsFor misidentifies some windows if TimeWindows#advanceBy is 
> used
> --
>
> Key: KAFKA-3784
> URL: https://issues.apache.org/jira/browse/KAFKA-3784
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.0.0
>Reporter: Tom Rybak
>Assignee: Tom Rybak
>Priority: Minor
>  Labels: architecture
> Fix For: 0.10.0.1
>
>
> Using a time window of size 6 minutes with a 5 minute advanceBy results in 
> some of the timestamped data being inserted into the previous overlapping 
> window even though the event's timestamp > that window's end time. 
> The fault lies in TimeWindows#windowsFor which does not check that all the 
> windows it's adding have an endTime > event's timestamp. 



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-3784) TimeWindows#windowsFor misidentifies some windows if TimeWindows#advanceBy is used

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-3784?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15314736#comment-15314736
 ] 

ASF GitHub Bot commented on KAFKA-3784:
---

Github user asfgit closed the pull request at:

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


> TimeWindows#windowsFor misidentifies some windows if TimeWindows#advanceBy is 
> used
> --
>
> Key: KAFKA-3784
> URL: https://issues.apache.org/jira/browse/KAFKA-3784
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.0.0
>Reporter: Tom Rybak
>Assignee: Tom Rybak
>Priority: Minor
>  Labels: architecture
> Fix For: 0.10.0.1
>
>
> Using a time window of size 6 minutes with a 5 minute advanceBy results in 
> some of the timestamped data being inserted into the previous overlapping 
> window even though the event's timestamp > that window's end time. 
> The fault lies in TimeWindows#windowsFor which does not check that all the 
> windows it's adding have an endTime > event's timestamp. 



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-3784) TimeWindows#windowsFor misidentifies some windows if TimeWindows#advanceBy is used

[Guozhang Wang (JIRA)]

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

Guozhang Wang updated KAFKA-3784:
-
Labels: architecture  (was: )

> TimeWindows#windowsFor misidentifies some windows if TimeWindows#advanceBy is 
> used
> --
>
> Key: KAFKA-3784
> URL: https://issues.apache.org/jira/browse/KAFKA-3784
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.0.0
>Reporter: Tom Rybak
>Assignee: Tom Rybak
>Priority: Minor
>  Labels: architecture
> Fix For: 0.10.0.1
>
>
> Using a time window of size 6 minutes with a 5 minute advanceBy results in 
> some of the timestamped data being inserted into the previous overlapping 
> window even though the event's timestamp > that window's end time. 
> The fault lies in TimeWindows#windowsFor which does not check that all the 
> windows it's adding have an endTime > event's timestamp. 



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] kafka pull request #1462: KAFKA-3784 TimeWindows#windowsFor misidentifies so...

[asfgit]

Github user asfgit closed the pull request at:

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


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (KAFKA-3775) Throttle maximum number of tasks assigned to a single KafkaStreams

[Jay Kreps (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-3775?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15314721#comment-15314721
 ] 

Jay Kreps commented on KAFKA-3775:
--

I'd have to agree I think this is a memory management problem I think we should 
solve that directly rather than having you throttle tasks which is a very 
roundabout way to control memory and would then lead to unprocessed partitions. 
I think the problems are
1. Kafka consumer does a poor job of controlling memory usage (known issue, 
needs to be fixed)
2. We may exacerbate it by giving configs around memory that are per-task when 
they should likely be global.
3. Maybe others?

> Throttle maximum number of tasks assigned to a single KafkaStreams
> --
>
> Key: KAFKA-3775
> URL: https://issues.apache.org/jira/browse/KAFKA-3775
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Affects Versions: 0.10.0.0
>Reporter: Yuto Kawamura
>Assignee: Yuto Kawamura
> Fix For: 0.10.1.0
>
>
> As of today, if I start a Kafka Streams app on a single machine which 
> consists of single KafkaStreams instance, that instance gets all partitions 
> of the target topic assigned.
> As we're using it to process topics which has huge number of partitions and 
> message traffic, it is a problem that we don't have a way of throttling the 
> maximum amount of partitions assigned to a single instance.
> In fact, when we started a Kafka Streams app which consumes a topic which has 
> more than 10MB/sec traffic of each partition we saw that all partitions 
> assigned to the first instance and soon the app dead by OOM.
> I know that there's some workarounds considerable here. for example:
> - Start multiple instances at once so the partitions distributed evenly.
>   => Maybe works. but as Kafka Streams is a library but not an execution 
> framework, there's no predefined procedure of starting Kafka Streams apps so 
> some users might wanna take an option to start the first single instance and 
> check if it works as expected with lesster number of partitions(I want :p)
> - Adjust config parameters such as {{buffered.records.per.partition}}, 
> {{max.partition.fetch.bytes}} and {{max.poll.records}} to reduce the heap 
> pressure.
>   => Maybe works. but still have two problems IMO:
>   - Still leads traffic explosion with high throughput processing as it 
> accepts all incoming messages from hundreads of partitions.
>   - In the first place, by the distributed system principle, it's wired that 
> users don't have a away to control maximum "partitions" assigned to a single 
> shard(an instance of KafkaStreams here). Users should be allowed to provide 
> the maximum amount of partitions that is considered as possible to be 
> processed with single instance(or host).
> Here, I'd like to introduce a new configuration parameter 
> {{max.tasks.assigned}}, which limits the number of tasks(a notion of 
> partition) assigned to the processId(which is the notion of single 
> KafkaStreams instance).
> At the same time we need to change StreamPartitionAssignor(TaskAssignor) to 
> tolerate the incomplete assignment. That is, Kafka Streams should continue 
> working for the part of partitions even there are some partitions left 
> unassigned, in order to satisfy this> "user may want to take an option to 
> start the first single instance and check if it works as expected with 
> lesster number of partitions(I want :p)".
> I've implemented the rough POC for this. PTAL and if it make sense I will 
> continue sophisticating it.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-3705) Support non-key joining in KTable

[Guozhang Wang (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-3705?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15314643#comment-15314643
 ] 

Guozhang Wang commented on KAFKA-3705:
--

Another thing we may want to consider is to use a different memtable option in 
RocksDB: 
https://github.com/facebook/rocksdb/wiki/Hash-based-memtable-implementations

> Support non-key joining in KTable
> -
>
> Key: KAFKA-3705
> URL: https://issues.apache.org/jira/browse/KAFKA-3705
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Reporter: Guozhang Wang
>Assignee: Liquan Pei
>  Labels: api
> Fix For: 0.10.1.0
>
>
> Today in Kafka Streams DSL, KTable joins are only based on keys. If users 
> want to join a KTable A by key {{a}} with another KTable B by key {{b}} but 
> with a "foreign key" {{a}}, and assuming they are read from two topics which 
> are partitioned on {{a}} and {{b}} respectively, they need to do the 
> following pattern:
> {code}
> tableB' = tableB.groupBy(/* select on field "a" */).agg(...); // now tableB' 
> is partitioned on "a"
> tableA.join(tableB', joiner);
> {code}
> Even if these two tables are read from two topics which are already 
> partitioned on {{a}}, users still need to do the pre-aggregation in order to 
> make the two joining streams to be on the same key. This is a draw-back from 
> programability and we should fix it.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-3775) Throttle maximum number of tasks assigned to a single KafkaStreams

[Guozhang Wang (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-3775?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15314606#comment-15314606
 ] 

Guozhang Wang commented on KAFKA-3775:
--

Hi [~mjsax] [~kawamuray], great discussion regarding 1) above! Personally I 
think it is flexible to let users specify which partitions among all the topics 
the topology defines as sources to be processed, which is better from a user 
experience point of view.

Actually it is not completely true that "users don't have a away to control 
maximum "partitions" assigned to a single shard(an instance of KafkaStreams 
here)." In fact, the user customizable {{PartitionGrouper}} is used exactly for 
that, which takes the list of all topic-partitions as input, generates the 
tasks with each task assigned with some topic-partitions. The 
{{DefaultPartitionGrouper}} of course tries to capture all topic-partitions and 
generates multiple tasks for them. But users can also customize it by, for 
example, generating only one task which takes one partition for each of the 
input topic, and this single task will be assigned to the ONLY instance in your 
case. NOTE that this partition grouper is global, such that if you have two 
instances, both of them will execute the same {{PartitionGrouper}}, and if only 
one task is generated, some instance will become completely idle, and this need 
to be communicated clearly to users. Does that sound good?

> Throttle maximum number of tasks assigned to a single KafkaStreams
> --
>
> Key: KAFKA-3775
> URL: https://issues.apache.org/jira/browse/KAFKA-3775
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Affects Versions: 0.10.0.0
>Reporter: Yuto Kawamura
>Assignee: Yuto Kawamura
> Fix For: 0.10.1.0
>
>
> As of today, if I start a Kafka Streams app on a single machine which 
> consists of single KafkaStreams instance, that instance gets all partitions 
> of the target topic assigned.
> As we're using it to process topics which has huge number of partitions and 
> message traffic, it is a problem that we don't have a way of throttling the 
> maximum amount of partitions assigned to a single instance.
> In fact, when we started a Kafka Streams app which consumes a topic which has 
> more than 10MB/sec traffic of each partition we saw that all partitions 
> assigned to the first instance and soon the app dead by OOM.
> I know that there's some workarounds considerable here. for example:
> - Start multiple instances at once so the partitions distributed evenly.
>   => Maybe works. but as Kafka Streams is a library but not an execution 
> framework, there's no predefined procedure of starting Kafka Streams apps so 
> some users might wanna take an option to start the first single instance and 
> check if it works as expected with lesster number of partitions(I want :p)
> - Adjust config parameters such as {{buffered.records.per.partition}}, 
> {{max.partition.fetch.bytes}} and {{max.poll.records}} to reduce the heap 
> pressure.
>   => Maybe works. but still have two problems IMO:
>   - Still leads traffic explosion with high throughput processing as it 
> accepts all incoming messages from hundreads of partitions.
>   - In the first place, by the distributed system principle, it's wired that 
> users don't have a away to control maximum "partitions" assigned to a single 
> shard(an instance of KafkaStreams here). Users should be allowed to provide 
> the maximum amount of partitions that is considered as possible to be 
> processed with single instance(or host).
> Here, I'd like to introduce a new configuration parameter 
> {{max.tasks.assigned}}, which limits the number of tasks(a notion of 
> partition) assigned to the processId(which is the notion of single 
> KafkaStreams instance).
> At the same time we need to change StreamPartitionAssignor(TaskAssignor) to 
> tolerate the incomplete assignment. That is, Kafka Streams should continue 
> working for the part of partitions even there are some partitions left 
> unassigned, in order to satisfy this> "user may want to take an option to 
> start the first single instance and check if it works as expected with 
> lesster number of partitions(I want :p)".
> I've implemented the rough POC for this. PTAL and if it make sense I will 
> continue sophisticating it.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Re: [DISCUSS] KIP-62: Allow consumer to send heartbeats from a background thread

[Jason Gustafson]

Hey Ewen,

I confess your comments caught me off guard. It never occurred to me that
anyone would ask for a rebalance timeout so that it could be set _larger_
than the process timeout. Even with buffered or batch processing, I would
usually expect flushing before a rebalance to take no more time than a
periodic flush. Otherwise, I'd probably try to see if there was some
workload I could push into periodic flushes so that rebalances could
complete faster. But supposing this isn't possible or practical in some
cases, I'm wondering how limiting it would be in practice to have only the
one timeout in this case? I'm a little reluctant to add the additional
timeout since I think most users would not have a strong need to keep a
tight bound on normal processing time. (I'm also reminded that Jay
mentioned he might have to dock everyone's pay 5% for each new timeout we
introduce ;-)

Thanks,
Jason




On Thu, Jun 2, 2016 at 7:30 PM, Guozhang Wang  wrote:

> Hi Ewen,
>
> I think you are right, the rebalance process could potentially involve all
> the delayed compute / IO. More specifically, this is what I think of the
> rebalance process:
>
> 1. Coordinator decides to rebalance, start ticking based on rebalance time
> out.
> 2. Consumer realize rebalance needed when calling poll(); trigger
> onPartitionRevoked().
> 3. Consumer sends JoinGroupRequest;
> 4. Coordinator send JoinGroupResponse; start ticking on the leader.
> 5. Leader compute and send SyncGroupRequest
> 6. Coordinator send SyncGroupResponse; start ticking on session timeout.
> 7. Consumer get new assignment; trigger onPartitionAssigned().
>
> In the above process: delayed compute / IO is usually done at step 2);
> workload initialization is usually done in step 7); and some admin work
> (like in Kafka Streams) are likely to be done in step 5). As in the current
> KIP proposal the rebalance timeout on the coordinator start ticking on 1)
> on everyone in the group, and stop ticking on 3); it start ticking on
> leader again on step 4), and stop upon step 5). In this case the delayed
> compute / IO contained in step 2) is covered by this rebalance timeout.
>
> That being said, I think for "worst case", the time of processing a single
> record would still be similar to rebalancing, since both of which could
> result in completing all delayed compute / IO so far. And since "processing
> timeout" is used to cover the worst case, it should be still OK?
>
>
> Guozhang
>
>
>
>
> On Thu, Jun 2, 2016 at 5:55 PM, Ewen Cheslack-Postava 
> wrote:
>
> > Jason,
> >
> > I've been thinking about this more in terms of something like Connect. I
> > think the rebalance timeout may be a bit different from the process
> > timeout, and even the process timeout is a bit of a misnomer.
> >
> > We sort of talk about the process timeout as if it can be an indicator of
> > maximum processing time for a record/batch. This makes sense for a case
> of
> > a data-dependent load (i.e. you can only load some data from slow storage
> > after seeing some data) where that load might be very large compared to
> > normal processing time. It also makes sense if you have auto commit
> enabled
> > because you need to be completely finished processing the data before
> > calling poll() again, so that time before you call another consumer API
> > actually reflects processing time.
> >
> > It might makes less sense in cases like streams (or any other app) that
> > batch writes to disk, or connectors that "process" a message by enqueuing
> > the data, but won't commit offsets until data is flushed, possibly during
> > some other, much later iteration of processing. In this case I think
> > processing time and rebalance time could potentially differ
> significantly.
> > During normal processing, you can potentially pipeline quite a bit,
> > buffering up changes, flushing as needed, but then only committing once
> > flushing is complete. But rebalancing is different then -- you *must*
> > finish flushing all the data or manually choose to discard the data
> > (presumably by doing something like watching for the process timeout you
> > set and bailing early, only committing the offsets for data you've
> > flushed). If you have lots of data built up, the cost for rebalancing
> could
> > be a *lot* higher than the maximum time you would otherwise see between
> > calls to consumer APIs to indicate processing progress.
> >
> > The thing that makes these cases different is that processing isn't
> > actually tied to calls to the consumer API. You can queue up / pipeline /
> > defer some of the work. (By the way, this is currently a limitation of
> sink
> > connectors that I'm not thrilled about -- offset commit requires a full
> > flush, whereas some coordination with the sink connector to not require a
> > full flush except on rebalances would be much nicer, albeit more
> difficult
> > for sink connectors to implement.)
> >
> > -Ewen
> >
> >
> >
> > On Thu, Jun 2, 2016 

[jira] [Updated] (KAFKA-3784) TimeWindows#windowsFor misidentifies some windows if TimeWindows#advanceBy is used

[Guozhang Wang (JIRA)]

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

Guozhang Wang updated KAFKA-3784:
-
Assignee: Tom Rybak

> TimeWindows#windowsFor misidentifies some windows if TimeWindows#advanceBy is 
> used
> --
>
> Key: KAFKA-3784
> URL: https://issues.apache.org/jira/browse/KAFKA-3784
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.0.0
>Reporter: Tom Rybak
>Assignee: Tom Rybak
>Priority: Minor
>
> Using a time window of size 6 minutes with a 5 minute advanceBy results in 
> some of the timestamped data being inserted into the previous overlapping 
> window even though the event's timestamp > that window's end time. 
> The fault lies in TimeWindows#windowsFor which does not check that all the 
> windows it's adding have an endTime > event's timestamp. 



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-3784) TimeWindows#windowsFor misidentifies some windows if TimeWindows#advanceBy is used

[Guozhang Wang (JIRA)]

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

Guozhang Wang updated KAFKA-3784:
-
Component/s: (was: core)
 streams

> TimeWindows#windowsFor misidentifies some windows if TimeWindows#advanceBy is 
> used
> --
>
> Key: KAFKA-3784
> URL: https://issues.apache.org/jira/browse/KAFKA-3784
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.0.0
>Reporter: Tom Rybak
>Assignee: Tom Rybak
>Priority: Minor
>
> Using a time window of size 6 minutes with a 5 minute advanceBy results in 
> some of the timestamped data being inserted into the previous overlapping 
> window even though the event's timestamp > that window's end time. 
> The fault lies in TimeWindows#windowsFor which does not check that all the 
> windows it's adding have an endTime > event's timestamp. 



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-3777) Extract the LRU cache out of RocksDBStore

[Guozhang Wang (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-3777?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15314561#comment-15314561
 ] 

Guozhang Wang commented on KAFKA-3777:
--

Adding to what Jay mentioned, we are already disabling the WAL logging of 
RocksDB in Kafka Streams.

> Extract the LRU cache out of RocksDBStore
> -
>
> Key: KAFKA-3777
> URL: https://issues.apache.org/jira/browse/KAFKA-3777
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Affects Versions: 0.10.1.0
>Reporter: Eno Thereska
> Fix For: 0.10.1.0
>
>
> The LRU cache that is currently inside the RocksDbStore class. As part of 
> KAFKA-3776 it needs to come outside of RocksDbStore and be a separate 
> component used in:
> 1. KGroupedStream.aggregate() / reduce(), 
> 2. KStream.aggregateByKey() / reduceByKey(),
> 3. KTable.to() (this will be done in KAFKA-3779).
> As all of the above operators can have a cache on top to deduplicate the 
> materialized state store in RocksDB.
> The scope of this JIRA is to extract out the cache of RocksDBStore, and keep 
> them as item 1) and 2) above; and it should be done together / after 
> KAFKA-3780.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Re: [DISCUSS] KIP-54 Sticky Partition Assignment Strategy

[Guozhang Wang]

Sorry for being late on this thread.

The assign() function is auto-triggered during the rebalance by one of the
consumers when it receives all subscription information collected from the
server-side coordinator.

More details can be found here:
https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Client-side+Assignment+Proposal#KafkaClient-sideAssignmentProposal-ConsumerEmbeddedProtocol

As for Kafka Streams, they way it did "stickiness" is by 1) let all
consumers put their current assigned topic-partitions and server ids into
the "metadata" field of the JoinGroupRequest, 2) when the selected consumer
triggers assign() along with all the subscriptions as well as their
metadata, it can parse the metadata to learn about the existing assignment
map; and hence when making the new assignment it will try to assign
partitions to its current owners "with best effort".


Hope this helps.


Guozhang


On Thu, May 26, 2016 at 4:56 PM, Vahid S Hashemian <
vahidhashem...@us.ibm.com> wrote:

> Hi Guozhang,
>
> I was looking at the implementation of StreamsPartitionAssignor through
> its unit tests and expected to find some tests that
> - verify stickiness by making at least two calls to the assign() method
> (so we check the second assign() call output preserves the assignments
> coming from the first assign() call output); or
> - start off by a preset assignment, call assign() after some subscription
> change, and verify the previous assignment are preserved.
> But none of the methods seem to do these. Did I overlook them, or
> stickiness is being tested in some other fashion?
>
> Also, if there is a high-level write-up about how this assignor works
> could you please point me to it? Thanks.
>
> Regards.
> --Vahid
>
>
>
>
> From:   Guozhang Wang 
> To: "dev@kafka.apache.org" 
> Date:   05/02/2016 10:34 AM
> Subject:Re: [DISCUSS] KIP-54 Sticky Partition Assignment Strategy
>
>
>
> Just FYI, the StreamsPartitionAssignor in Kafka Streams are already doing
> some sort of sticky partitioning mechanism. This is done through the
> userData field though; i.e. all group members send their current "assigned
> partitions" in their join group request, which will be grouped and send to
> the leader, the leader then does best-effort for sticky-partitioning.
>
>
> Guozhang
>
> On Fri, Apr 29, 2016 at 9:48 PM, Ewen Cheslack-Postava 
> wrote:
>
> > I think I'm unclear how we leverage the
> > onPartitionsRevoked/onPartitionsAssigned here in any way that's
> different
> > from our normal usage -- certainly you can use them to generate a diff,
> but
> > you still need to commit when partitions are revoked and that has a
> > non-trivial cost. Are we just saying that you might be able to save some
> > overhead, e.g. closing/reopening some other resources by doing a flush
> but
> > not a close() or something? You still need to flush any output and
> commit
> > offsets before returning from onPartitionsRevoked, right? Otherwise you
> > couldn't guarantee clean handoff of partitions.
> >
> > In terms of the rebalancing, the basic requirements in the KIP seem
> sound.
> > Passing previous assignment data via UserData also seems reasonable
> since
> > it avoids redistributing all assignment data to all members and doesn't
> > rely on the next generation leader being a member of the current
> > generation. Hopefully this shouldn't be surprising since I think I
> > discussed this w/ Jason before he updated the relevant wiki pages :)
> >
> > -Ewen
> >
> >
> > On Mon, Apr 18, 2016 at 9:34 AM, Vahid S Hashemian <
> > vahidhashem...@us.ibm.com> wrote:
> >
> > > HI Jason,
> > >
> > > Thanks for your feedback.
> > >
> > > I believe your suggestion on how to take advantage of this assignor is
> > > valid. We can leverage onPartitionsRevoked() and
> onPartitionsAssigned()
> > > callbacks and do a comparison of assigned partitions before and after
> the
> > > re-balance and do the cleanup only if there is a change (e.g., if some
> > > previously assigned partition is not in the assignment).
> > >
> > > On your second question, a number of tests that I ran shows that the
> old
> > > assignments are preserved in the current implementation; except for
> when
> > > the consumer group leader is killed; in which case, a fresh assignment
> is
> > > performed. This is something that needs to be fixed. I tried to use
> your
> > > pointers to find out where the best place is to preserve the old
> > > assignment in such circumstances but have not been able to pinpoint
> it.
> > If
> > > you have any suggestion on this please share. Thanks.
> > >
> > > Regards,
> > > Vahid Hashemian
> > >
> > >
> > >
> > >
> > > From:   Jason Gustafson 
> > > To: dev@kafka.apache.org
> > > Date:   04/14/2016 11:37 AM
> > > Subject:Re: [DISCUSS] KIP-54 Sticky Partition Assignment
> Strategy
> > >
> > >
> > >
> > > Hi Vahid,
> > >
> > > Thanks for the proposal. I think one of 

Re: [DISCUSS] KIP-47 - Add timestamp-based log deletion policy

[Magnus Edenhill]

Bumping this thread so Wes can reply to it. Ignore this mail.

2016-02-24 0:36 GMT+01:00 Joel Koshy :

> Great - thanks for clarifying.
>
> Joel
>
> On Tue, Feb 23, 2016 at 1:47 PM, Bill Warshaw  wrote:
>
> > Sorry that I didn't see this comment before the meeting Joel.  I'll try
> to
> > clarify what I said at the meeting:
> >
> > - The KIP currently states that timestamp-based log deletion will only
> work
> > with LogAppendTime.  I need to update the KIP to reflect that, after the
> > work is done for KIP-33, it will work with both LogAppendTime and
> > CreateTime.
> > - To use the existing time-based retention mechanism to delete a precise
> > range of messages, a client application would need to do the following:
> >   - by default, turn off these retention mechanisms
> >   - when the application wishes to delete a range of messages which were
> > sent before a certain time, compute an approximate value to set
> > "log.retention.minutes" to, to create a window of messages based on that
> > timestamp that are ok to delete.  There is some degree of imprecision
> > implied here.
> >   - wait until we are confident that the log retention mechanism has been
> > run and deleted any stale segments
> >   - reset "log.retention.minutes" to turn off time-based log retention
> > until the next time the client application wants to delete something
> >
> > - To use the proposed timestamp-based retention mechanism, there is only
> > one step: the application just has to set "log.retention.min.timestamp"
> to
> > whatever time boundary it deems fit.  It doesn't need to compute any
> fuzzy
> > windows, try to wait until asynchronous processes have been completed or
> > continually flip settings between enabled and disabled.
> >
> > I will update the KIP to reflect the discussion around LogAppendTime vs
> > CreateTime and the work being done in KIP-33.
> >
> > Thanks,
> > Bill
> >
> >
> > On Tue, Feb 23, 2016 at 1:22 PM, Joel Koshy  wrote:
> >
> > > I'm having some trouble reconciling the current proposal with your
> > original
> > > requirement which was essentially being able to purge log data up to a
> > > precise point (an offset). The KIP currently suggests that
> > timestamp-based
> > > deletion would only work with LogAppendTime, so it does not seem
> > > significantly different from time-based retention (after KIP-32/33) -
> IOW
> > > to me it appears that you would need to use CreateTime and not
> > > LogAppendTime. Also one of the rejected alternatives observes that
> > changing
> > > the existing configuration settings to try to flush ranges of a given
> > > partition's log are problematic, but it seems to me you would have to
> do
> > > this in with timestamp-based deletion as well right? I think it would
> be
> > > useful for me if you or anyone else can go over the exact
> > > mechanics/workflow for accomplishing precise purges at today's KIP
> > meeting.
> > >
> > > Thanks,
> > >
> > > Joel
> > >
> > > On Monday, February 22, 2016, Bill Warshaw 
> wrote:
> > >
> > > > Sounds good.  I'll hold off on sending out a VOTE thread until after
> > the
> > > > KIP meeting tomorrow.
> > > >
> > > > On Mon, Feb 22, 2016 at 12:56 PM, Becket Qin 
> > > wrote:
> > > >
> > > > > Hi Jun,
> > > > >
> > > > > I think it makes sense to implement KIP-47 after KIP-33 so we can
> > make
> > > it
> > > > > work for both LogAppendTime and CreateTime.
> > > > >
> > > > > And yes, I'm actively working on KIP-33. I had a voting thread on
> > > KIP-33
> > > > > before and I'll bump it up.
> > > > >
> > > > > Thanks,
> > > > >
> > > > > Jiangjie (Becket) Qin
> > > > >
> > > > >
> > > > >
> > > > > On Mon, Feb 22, 2016 at 9:11 AM, Jun Rao  wrote:
> > > > >
> > > > > > Becket,
> > > > > >
> > > > > > Since you submitted KIP-33, are you actively working on that? If
> > so,
> > > it
> > > > > > would make sense to implement KIP-47 after KIP-33 so that it
> works
> > > for
> > > > > both
> > > > > > CreateTime and LogAppendTime.
> > > > > >
> > > > > > Thanks,
> > > > > >
> > > > > > Jun
> > > > > >
> > > > > >
> > > > > >
> > > > > >
> > > > > > On Fri, Feb 19, 2016 at 6:25 PM, Bill Warshaw <
> wdwars...@gmail.com
> > >
> > > > > wrote:
> > > > > >
> > > > > > > Hi Jun,
> > > > > > >
> > > > > > > 1.  I thought more about Andrew's comment about LogAppendTime.
> > The
> > > > > > > time-based index you are referring to is associated with
> KIP-33,
> > > > > correct?
> > > > > > > Currently my implementation is just checking the last message
> in
> > a
> > > > > > segment,
> > > > > > > so we're restricted to LogAppendTime.  When the work for KIP-33
> > is
> > > > > > > completed, it sounds like CreateTime would also be valid.  Do
> you
> > > > > happen
> > > > > > to
> > > > > > > know if anyone is currently working on KIP-33?
> > > > > > >
> > > > > > > 2. I did update the wiki after reading your original 

Jenkins build is back to normal : kafka-trunk-jdk7 #1337

[Apache Jenkins Server]

See 

[jira] [Commented] (KAFKA-3790) Default options when removing ACLs do not comply with documentation

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-3790?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15314474#comment-15314474
 ] 

ASF GitHub Bot commented on KAFKA-3790:
---

GitHub user slaunay opened a pull request:

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

KAFKA-3790: Allow for removal of non specific ACLs

- remove ACLs with `aclMatch()` rather than `Object#equals(Object)`
- remove unused session argument from `aclMatch()` to reuse it in 
`removeAcls()`
- update test case for ACL removal management
- change test method `changeAclAndVerify(...)` to use an expected `Set` of 
ACLs rather than relying on `Object#equals(Object)`

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/slaunay/kafka 
bugfix/KAFKA-3790-remove-acls-matching-wildcard

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/1468.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #1468


commit cd2934579ae49afad92eda5f6e36069e2f1d63d4
Author: Sebastien Launay 
Date:   2016-06-02T20:29:11Z

KAFKA-3790: Allow for removal of non specific ACLs

- remove ACLs with aclMatch() rather than Object#equals(Object)
- remove unused session argument from aclMatch() to reuse it in
  removeAcls()
- update test case for ACL removal management
- change test method changeAclAndVerify(...) to use an expected Set of
  ACLs rather than relying on Object#equals(Object)




> Default options when removing ACLs do not comply with documentation
> ---
>
> Key: KAFKA-3790
> URL: https://issues.apache.org/jira/browse/KAFKA-3790
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.9.0.1, 0.10.0.0
>Reporter: Sébastien Launay
>Priority: Minor
>
> When removing ACLs without providing options like principal, host or 
> operation, we got a prompt for removing all the matching ACLs but when 
> executing the command none get removed.
> The following commands can be used to reproduce the inconsistency:
> {noformat}
> $ ./bin/kafka-acls.sh --authorizer-properties 
> zookeeper.connect=localhost:2181 -list -topic test
> Current ACLs for resource `Topic:test`: 
> $ ./bin/kafka-acls.sh --authorizer-properties 
> zookeeper.connect=localhost:2181 --add --allow-principal User:Alice 
> --operation Write --topic test --allow-host 1.2.3.4
> Adding ACLs for resource `Topic:test`: 
>   User:Alice has Allow permission for operations: Write from hosts: 
> 1.2.3.4 
> Current ACLs for resource `Topic:test`: 
>   User:Alice has Allow permission for operations: Write from hosts: 
> 1.2.3.4 
> $ ./bin/kafka-acls.sh --authorizer-properties 
> zookeeper.connect=localhost:2181 --remove --allow-principal User:Alice 
> --topic test 
> Are you sure you want to remove ACLs: 
>   User:Alice has Allow permission for operations: All from hosts: * 
>  from resource `Topic:test`? (y/n)
> y
> Current ACLs for resource `Topic:test`: 
>   User:Alice has Allow permission for operations: Write from hosts: 
> 1.2.3.4 
> {noformat}
> *The Current ACLs for resource {{Topic:test}} is expected to be empty after 
> the last command.*
> Only a specific ACL (when all options mentioned above are provided) or else 
> all the ACLs for a given resource (none of the options mentioned above are 
> provided) can get removed as shown by the following code snippets:
> {noformat}
>   // AclCommand.scala
>   ...
>   private def removeAcl(opts: AclCommandOptions) {
> withAuthorizer(opts) { authorizer =>
>   val resourceToAcl = getResourceToAcls(opts)
>   for ((resource, acls) <- resourceToAcl) {
> if (acls.isEmpty) {
>   if (confirmAction(opts, s"Are you sure you want to delete all ACLs 
> for resource `${resource}`? (y/n)"))
> authorizer.removeAcls(resource)
> } else {
>   if (confirmAction(opts, s"Are you sure you want to remove ACLs: 
> $Newline ${acls.map("\t" + _).mkString(Newline)} $Newline from resource 
> `${resource}`? (y/n)"))
> authorizer.removeAcls(acls, resource)
> }
>   }
>   listAcl(opts)
> }
>   }
> ...
>   // SimpleAclAuthorizer.scala
> ...
>   override def removeAcls(aclsTobeRemoved: Set[Acl], resource: Resource): 
> Boolean = {
>  inWriteLock(lock) {
>updateResourceAcls(resource) { currentAcls =>
> currentAcls -- aclsTobeRemoved
>}
>  }
>}
> {noformat}
> A workaround consists of listing the ACL in order to know which exact one to 
> remove which make the automation of ACL management trickier.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] kafka pull request #1468: KAFKA-3790: Allow for removal of non specific ACLs

[slaunay]

GitHub user slaunay opened a pull request:

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

KAFKA-3790: Allow for removal of non specific ACLs

- remove ACLs with `aclMatch()` rather than `Object#equals(Object)`
- remove unused session argument from `aclMatch()` to reuse it in 
`removeAcls()`
- update test case for ACL removal management
- change test method `changeAclAndVerify(...)` to use an expected `Set` of 
ACLs rather than relying on `Object#equals(Object)`

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/slaunay/kafka 
bugfix/KAFKA-3790-remove-acls-matching-wildcard

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/1468.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #1468


commit cd2934579ae49afad92eda5f6e36069e2f1d63d4
Author: Sebastien Launay 
Date:   2016-06-02T20:29:11Z

KAFKA-3790: Allow for removal of non specific ACLs

- remove ACLs with aclMatch() rather than Object#equals(Object)
- remove unused session argument from aclMatch() to reuse it in
  removeAcls()
- update test case for ACL removal management
- change test method changeAclAndVerify(...) to use an expected Set of
  ACLs rather than relying on Object#equals(Object)




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Updated] (KAFKA-3789) Upgrade Snappy to fix snappy decompression errors

[Grant Henke (JIRA)]

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

Grant Henke updated KAFKA-3789:
---
Status: Patch Available  (was: Open)

> Upgrade Snappy to fix snappy decompression errors
> -
>
> Key: KAFKA-3789
> URL: https://issues.apache.org/jira/browse/KAFKA-3789
> Project: Kafka
>  Issue Type: Bug
>  Components: core
>Affects Versions: 0.10.0.0
>Reporter: Grant Henke
>Assignee: Grant Henke
>Priority: Critical
> Fix For: 0.10.0.1
>
>
> snappy-java recently fixed a bug where parsing the MAGIC HEADER was being 
> handled incorrectly: https://github.com/xerial/snappy-java/issues/142
> This issue caused "unknown broker exceptions" in the clients and prevented 
> these messages from being appended to the log when messages were written 
> using snappy c bindings in clients like librdkafka or ruby-kafka and read 
> using snappy-java in the broker.   
> The related librdkafka issue is here: 
> https://github.com/edenhill/librdkafka/issues/645
> I am able to regularly reproduce the issue with librdkafka in 0.10 and after 
> upgrading snappy-java to 1.1.2.6 the issue is resolved. 



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-3789) Upgrade Snappy to fix snappy decompression errors

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-3789?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15314443#comment-15314443
 ] 

ASF GitHub Bot commented on KAFKA-3789:
---

GitHub user granthenke opened a pull request:

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

KAFKA-3789: Upgrade Snappy to fix snappy decompression errors



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/granthenke/kafka snappy-fix

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/1467.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #1467


commit f08e12a9a59fb143049f337bccdb1b9dc21120d1
Author: Grant Henke 
Date:   2016-06-03T17:13:23Z

KAFKA-3789: Upgrade Snappy to fix snappy decompression errors




> Upgrade Snappy to fix snappy decompression errors
> -
>
> Key: KAFKA-3789
> URL: https://issues.apache.org/jira/browse/KAFKA-3789
> Project: Kafka
>  Issue Type: Bug
>  Components: core
>Affects Versions: 0.10.0.0
>Reporter: Grant Henke
>Assignee: Grant Henke
>Priority: Critical
> Fix For: 0.10.0.1
>
>
> snappy-java recently fixed a bug where parsing the MAGIC HEADER was being 
> handled incorrectly: https://github.com/xerial/snappy-java/issues/142
> This issue caused "unknown broker exceptions" in the clients and prevented 
> these messages from being appended to the log when messages were written 
> using snappy c bindings in clients like librdkafka or ruby-kafka and read 
> using snappy-java in the broker.   
> The related librdkafka issue is here: 
> https://github.com/edenhill/librdkafka/issues/645
> I am able to regularly reproduce the issue with librdkafka in 0.10 and after 
> upgrading snappy-java to 1.1.2.6 the issue is resolved. 



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] kafka pull request #1467: KAFKA-3789: Upgrade Snappy to fix snappy decompres...

[granthenke]

GitHub user granthenke opened a pull request:

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

KAFKA-3789: Upgrade Snappy to fix snappy decompression errors



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/granthenke/kafka snappy-fix

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/1467.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #1467


commit f08e12a9a59fb143049f337bccdb1b9dc21120d1
Author: Grant Henke 
Date:   2016-06-03T17:13:23Z

KAFKA-3789: Upgrade Snappy to fix snappy decompression errors




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

Re: KStreams Rewind Offset

[Guozhang Wang]

Hello Michael,

Just want to clarify a few thing before I can provide some suggestions in
simplifying your pipeline with Kafka Streams: as for the "race condition
between reading the stream data and the data being removed", do you mean
that the stream records is actually mutable, hence after they are read,
their content could be modified? For example:

stream at time 0:   (begin), (edge1),
(edge2), ... (end)

graphDB read up to (edge2) at time 1: (begin), (edge1), (edge2), ...
(end)

stream gets modified at time 0:(begin), *(edge1: deleted)*,
(edge2), ... (end)


If that is the case, I would suggest you make the stream records immutable,
but append "removal" records to the end of the stream when removal happens,
which matches better with the stream processing principles, for example:

(edge1: created), (edge2: created), (edge1: removed), ...

So a later "removed" record can be treated just as an update to a previous
record with the same key "edge1", and upon reading it the graphDB can just
cleanup the created edge accordingly.


If there are other race causes that prevent you to make immutable live
streams, please let me know and we can discuss further.


Guozhang




On Thu, Jun 2, 2016 at 6:55 AM, Michael D. Coon 
wrote:

> Mattias,
>That's disappointing given that Kafka offers me the ability to rewind
> and replay data. My use case is that we are building graph data structures
> based on data indexed from a live stream. At any time, the live data
> content may be marked for deletion for any number of reasons; but during
> that marking process if a graph structure is being built, it may not
> realize the data was marked for deletion (i.e. there is a race between
> graph referencing the data and the data being removed).
>
>We need to be able to subsequently go back and clean up the graph data
> once we realize the graph contains data that was marked for deletion. But
> we can't delete/cleanup the graph until it completes...so we thought we
> could track all data referenced by the graph being created and once it was
> complete, subsequently replay the data references and determine if any were
> marked for removal and subsequently clean up the graph. We hoped that by
> sending "start/end" indicators into a graph data reference topic, some
> KStreams flow could see the "end", recognize that the graph completed, and
> simply replay all its data references to cleanup the graph. I guess we
> could use a standard consumer and do this outside of KStreams. Not a big
> deal...was just hoping to keep things in the KStreams realm. I'm sure there
> are other ways to solve this even outside of using Kafka at all; but why do
> that? :)
> Mike
>
>
>
> On Thursday, June 2, 2016 8:59 AM, Matthias J. Sax <
> matth...@confluent.io> wrote:
>
>
>  Hi Mike,
>
> currently, this is not possible. We are already discussing some changes
> with regard to reprocess. However, I doubt that going back to a specific
> offset of a specific partition will be supported as it would be too
> difficult to reset the internal data structures and intermediate results
> correctly (also with regard to committing)
>
> What is your exact use case? What kind of feature are you looking for?
> We are always interested to get feedback/idea from users.
>
>
> -Matthias
>
> On 06/01/2016 08:21 PM, Michael D. Coon wrote:
> > All,
> >  I think it's great that the ProcessorContext offers the partition and
> offset of the current record being processed; however, it offers no way for
> me to actually use the information. I would like to be able to rewind to a
> particular offset on a partition if I needed to. The consumer is also not
> exposed to me so I couldn't access things directly that way either. Is this
> in the works or would it interfere with rebalancing/auto-commits?
> > Mike
> >
> >
>
>
>
>



-- 
-- Guozhang

[jira] [Created] (KAFKA-3790) Default options when removing ACLs do not comply with documentation

[JIRA]

Sébastien Launay created KAFKA-3790:
---

 Summary: Default options when removing ACLs do not comply with 
documentation
 Key: KAFKA-3790
 URL: https://issues.apache.org/jira/browse/KAFKA-3790
 Project: Kafka
  Issue Type: Bug
Affects Versions: 0.10.0.0, 0.9.0.1
Reporter: Sébastien Launay
Priority: Minor


When removing ACLs without providing options like principal, host or operation, 
we got a prompt for removing all the matching ACLs but when executing the 
command none get removed.

The following commands can be used to reproduce the inconsistency:
{noformat}
$ ./bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 
-list -topic test
Current ACLs for resource `Topic:test`: 


$ ./bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 
--add --allow-principal User:Alice --operation Write --topic test --allow-host 
1.2.3.4
Adding ACLs for resource `Topic:test`: 
User:Alice has Allow permission for operations: Write from hosts: 
1.2.3.4 

Current ACLs for resource `Topic:test`: 
User:Alice has Allow permission for operations: Write from hosts: 
1.2.3.4 

$ ./bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 
--remove --allow-principal User:Alice --topic test 
Are you sure you want to remove ACLs: 
User:Alice has Allow permission for operations: All from hosts: * 
 from resource `Topic:test`? (y/n)
y
Current ACLs for resource `Topic:test`: 
User:Alice has Allow permission for operations: Write from hosts: 
1.2.3.4 

{noformat}

*The Current ACLs for resource {{Topic:test}} is expected to be empty after the 
last command.*

Only a specific ACL (when all options mentioned above are provided) or else all 
the ACLs for a given resource (none of the options mentioned above are 
provided) can get removed as shown by the following code snippets:
{noformat}
  // AclCommand.scala
  ...
  private def removeAcl(opts: AclCommandOptions) {
withAuthorizer(opts) { authorizer =>
  val resourceToAcl = getResourceToAcls(opts)

  for ((resource, acls) <- resourceToAcl) {
if (acls.isEmpty) {
  if (confirmAction(opts, s"Are you sure you want to delete all ACLs 
for resource `${resource}`? (y/n)"))
authorizer.removeAcls(resource)
} else {
  if (confirmAction(opts, s"Are you sure you want to remove ACLs: 
$Newline ${acls.map("\t" + _).mkString(Newline)} $Newline from resource 
`${resource}`? (y/n)"))
authorizer.removeAcls(acls, resource)
}
  }

  listAcl(opts)
}
  }
...
  // SimpleAclAuthorizer.scala
...
  override def removeAcls(aclsTobeRemoved: Set[Acl], resource: Resource): 
Boolean = {
 inWriteLock(lock) {
   updateResourceAcls(resource) { currentAcls =>
currentAcls -- aclsTobeRemoved
   }
 }
   }
{noformat}

A workaround consists of listing the ACL in order to know which exact one to 
remove which make the automation of ACL management trickier.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

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

[Apache Jenkins Server]

See 

Changes:

[me] MINOR: set charset of Javadoc to UTF-8

--
Started by an SCM change
[EnvInject] - Loading node environment variables.
Building remotely on ubuntu3 (Ubuntu ubuntu legacy-ubuntu) in workspace 

 > git rev-parse --is-inside-work-tree # timeout=10
Fetching changes from the remote Git repository
 > git config remote.origin.url 
 > https://git-wip-us.apache.org/repos/asf/kafka.git # timeout=10
Fetching upstream changes from https://git-wip-us.apache.org/repos/asf/kafka.git
 > git --version # timeout=10
 > git -c core.askpass=true fetch --tags --progress 
 > https://git-wip-us.apache.org/repos/asf/kafka.git 
 > +refs/heads/*:refs/remotes/origin/*
 > git rev-parse refs/remotes/origin/trunk^{commit} # timeout=10
 > git rev-parse refs/remotes/origin/origin/trunk^{commit} # timeout=10
Checking out Revision a27030693bd89b16a436ba873192ebc98c4c7910 
(refs/remotes/origin/trunk)
 > git config core.sparsecheckout # timeout=10
 > git checkout -f a27030693bd89b16a436ba873192ebc98c4c7910
 > git rev-list 76ca6f8cc1a6533866a2bd7ca6a7d32314d21dba # timeout=10
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK1_8_0_66_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk1.8.0_66
[kafka-trunk-jdk8] $ /bin/bash -xe /tmp/hudson6413433135867037573.sh
+ 
/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2/bin/gradle
To honour the JVM settings for this build a new JVM will be forked. Please 
consider using the daemon: 
http://gradle.org/docs/2.4-rc-2/userguide/gradle_daemon.html.
Building project 'core' with Scala version 2.10.6
:downloadWrapper

BUILD SUCCESSFUL

Total time: 12.736 secs
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK1_8_0_66_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk1.8.0_66
[kafka-trunk-jdk8] $ /bin/bash -xe /tmp/hudson2027820784848344228.sh
+ export GRADLE_OPTS=-Xmx1024m
+ GRADLE_OPTS=-Xmx1024m
+ ./gradlew -Dorg.gradle.project.maxParallelForks=1 
-Dorg.gradle.project.testLoggingEvents=started,passed,skipped,failed clean 
jarAll testAll
To honour the JVM settings for this build a new JVM will be forked. Please 
consider using the daemon: 
https://docs.gradle.org/2.13/userguide/gradle_daemon.html.
Building project 'core' with Scala version 2.10.6
Build file '/x1/jenkins/jenkins-slave/workspace/kafka-trunk-jdk8/build.gradle': 
line 239
useAnt has been deprecated and is scheduled to be removed in Gradle 3.0. The 
Ant-Based Scala compiler is deprecated, please see 
https://docs.gradle.org/current/userguide/scala_plugin.html.
:clean UP-TO-DATE
:clients:clean
:connect:clean UP-TO-DATE
:core:clean
:examples:clean UP-TO-DATE
:log4j-appender:clean UP-TO-DATE
:streams:clean UP-TO-DATE
:tools:clean UP-TO-DATE
:connect:api:clean UP-TO-DATE
:connect:file:clean UP-TO-DATE
:connect:json:clean UP-TO-DATE
:connect:runtime:clean UP-TO-DATE
:streams:examples:clean UP-TO-DATE
:jar_core_2_10
Building project 'core' with Scala version 2.10.6
:kafka-trunk-jdk8:clients:compileJavawarning: [options] bootstrap class path 
not set in conjunction with -source 1.7
Note: Some input files use unchecked or unsafe operations.
Note: Recompile with -Xlint:unchecked for details.
1 warning

:kafka-trunk-jdk8:clients:processResources UP-TO-DATE
:kafka-trunk-jdk8:clients:classes
:kafka-trunk-jdk8:clients:determineCommitId UP-TO-DATE
:kafka-trunk-jdk8:clients:createVersionFile
:kafka-trunk-jdk8:clients:jar
:jar_core_2_10 FAILED

FAILURE: Build failed with an exception.

* What went wrong:
org.gradle.api.internal.changedetection.state.FileCollectionSnapshotImpl cannot 
be cast to 
org.gradle.api.internal.changedetection.state.OutputFilesCollectionSnapshotter$OutputFilesSnapshot

* Try:
Run with --stacktrace option to get the stack trace. Run with --info or --debug 
option to get more log output.

BUILD FAILED

Total time: 13.769 secs
Build step 'Execute shell' marked build as failure
Recording test results
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK1_8_0_66_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk1.8.0_66
ERROR: Step ?Publish JUnit test result report? failed: No test report files 
were found. Configuration error?
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK1_8_0_66_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk1.8.0_66

[jira] [Created] (KAFKA-3789) Upgrade Snappy to fix snappy decompression errors

[Grant Henke (JIRA)]

Grant Henke created KAFKA-3789:
--

 Summary: Upgrade Snappy to fix snappy decompression errors
 Key: KAFKA-3789
 URL: https://issues.apache.org/jira/browse/KAFKA-3789
 Project: Kafka
  Issue Type: Bug
  Components: core
Affects Versions: 0.10.0.0
Reporter: Grant Henke
Assignee: Grant Henke
Priority: Critical
 Fix For: 0.10.0.1


snappy-java recently fixed a bug where parsing the MAGIC HEADER was being 
handled incorrectly: https://github.com/xerial/snappy-java/issues/142

This issue caused "unknown broker exceptions" in the clients and prevented 
these messages from being appended to the log when messages were written using 
snappy c bindings in clients like librdkafka or ruby-kafka and read using 
snappy-java in the broker.   

The related librdkafka issue is here: 
https://github.com/edenhill/librdkafka/issues/645

I am able to regularly reproduce the issue with librdkafka in 0.10 and after 
upgrading snappy-java to 1.1.2.6 the issue is resolved. 



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-3680) Make Java client classloading more flexible

[Avi Flax (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-3680?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15314411#comment-15314411
 ] 

Avi Flax commented on KAFKA-3680:
-

I’m very much looking forward to this fix! I’m using Kafka clients and Kafka 
Streams with JRuby — which works great with a “normal” JRuby app wherein you 
just run the code, but does not work once a JRuby project is packaged up into 
an “uber” JAR with a custom classloader. (Full details in [this 
issue|https://github.com/jruby/warbler/issues/381].)

> Make Java client classloading more flexible
> ---
>
> Key: KAFKA-3680
> URL: https://issues.apache.org/jira/browse/KAFKA-3680
> Project: Kafka
>  Issue Type: Improvement
>  Components: clients
>Affects Versions: 0.10.0.0
>Reporter: Rajini Sivaram
>Assignee: Rajini Sivaram
>
> JIRA corresponding to 
> [KIP-60|https://cwiki.apache.org/confluence/display/KAFKA/KIP-60+-+Make+Java+client+classloading+more+flexible]
>  to enable classloading of default classes and custom classes to work in 
> different classloading environments.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] kafka pull request #1413: MINOR: set charset of Javadoc to UTF-8

[asfgit]

Github user asfgit closed the pull request at:

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


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Created] (KAFKA-3788) Potential message lost when switching to new segment

[Arkadiusz Firus (JIRA)]

Arkadiusz Firus created KAFKA-3788:
--

 Summary: Potential message lost when switching to new segment
 Key: KAFKA-3788
 URL: https://issues.apache.org/jira/browse/KAFKA-3788
 Project: Kafka
  Issue Type: Bug
  Components: log
Affects Versions: 0.10.0.0, 0.9.0.1, 0.9.0.0
Reporter: Arkadiusz Firus
Assignee: Jay Kreps
Priority: Minor


If a new segment is needed method roll() from class kafka.log.Log is invoked. 
It prepares new segment and schedules _asynchronous_ flush of the previous 
segment.

Asynchronous call can lead to a problematic situation. As far as I know neither 
Linux nor Windows guarantees that the order of files persisted to disk will be 
the same as the order of writes to files. This means that records from the new 
segment can be flushed before the old ones which in case of power outage can 
lead to gaps between records.

Changing asynchronous invocation to synchronous one will solve the problem 
because we have guarantee that all records from the previous segment will be 
persisted to hard drive before we write any record to the new segment.

I am guessing that asynchronous invocation was chosen to increase performance 
but switching between segments is not so often. So it is not a big gain.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Build failed in Jenkins: kafka-0.10.0-jdk7 #114

[Apache Jenkins Server]

See 

Changes:

[ismael] MINOR: Add user overridden test logging events

--
Started by an SCM change
[EnvInject] - Loading node environment variables.
Building remotely on ubuntu-us1 (Ubuntu ubuntu ubuntu-us golang-ppa) in 
workspace 
 > git rev-parse --is-inside-work-tree # timeout=10
Fetching changes from the remote Git repository
 > git config remote.origin.url 
 > https://git-wip-us.apache.org/repos/asf/kafka.git # timeout=10
Fetching upstream changes from https://git-wip-us.apache.org/repos/asf/kafka.git
 > git --version # timeout=10
 > git -c core.askpass=true fetch --tags --progress 
 > https://git-wip-us.apache.org/repos/asf/kafka.git 
 > +refs/heads/*:refs/remotes/origin/*
 > git rev-parse refs/remotes/origin/0.10.0^{commit} # timeout=10
 > git rev-parse refs/remotes/origin/origin/0.10.0^{commit} # timeout=10
Checking out Revision 419e6517c75822d94974628151d399ae93e4a9f0 
(refs/remotes/origin/0.10.0)
 > git config core.sparsecheckout # timeout=10
 > git checkout -f 419e6517c75822d94974628151d399ae93e4a9f0
 > git rev-list c91f742deb8b83d59847f2afb7a786f18b34eb1d # timeout=10
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK_1_7U51_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk-1.7u51
[kafka-0.10.0-jdk7] $ /bin/bash -xe /tmp/hudson1569588920671295590.sh
+ 
/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2/bin/gradle
To honour the JVM settings for this build a new JVM will be forked. Please 
consider using the daemon: 
http://gradle.org/docs/2.4-rc-2/userguide/gradle_daemon.html.
Building project 'core' with Scala version 2.10.6
:downloadWrapper

BUILD SUCCESSFUL

Total time: 22.338 secs
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK_1_7U51_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk-1.7u51
[kafka-0.10.0-jdk7] $ /bin/bash -xe /tmp/hudson6677823416654567420.sh
+ export GRADLE_OPTS=-Xmx1024m
+ GRADLE_OPTS=-Xmx1024m
+ ./gradlew -Dorg.gradle.project.maxParallelForks=1 clean jarAll testAll
To honour the JVM settings for this build a new JVM will be forked. Please 
consider using the daemon: 
https://docs.gradle.org/2.13/userguide/gradle_daemon.html.
Building project 'core' with Scala version 2.10.6
Build file ': 
line 233
useAnt has been deprecated and is scheduled to be removed in Gradle 3.0. The 
Ant-Based Scala compiler is deprecated, please see 
https://docs.gradle.org/current/userguide/scala_plugin.html.
:clean UP-TO-DATE
:clients:clean
:connect:clean UP-TO-DATE
:core:clean
:examples:clean UP-TO-DATE
:log4j-appender:clean UP-TO-DATE
:streams:clean UP-TO-DATE
:tools:clean UP-TO-DATE
:connect:api:clean UP-TO-DATE
:connect:file:clean UP-TO-DATE
:connect:json:clean UP-TO-DATE
:connect:runtime:clean UP-TO-DATE
:streams:examples:clean UP-TO-DATE
:jar_core_2_10
Building project 'core' with Scala version 2.10.6
:kafka-0.10.0-jdk7:clients:compileJavaNote: Some input files use unchecked or 
unsafe operations.
Note: Recompile with -Xlint:unchecked for details.

:kafka-0.10.0-jdk7:clients:processResources UP-TO-DATE
:kafka-0.10.0-jdk7:clients:classes
:kafka-0.10.0-jdk7:clients:determineCommitId UP-TO-DATE
:kafka-0.10.0-jdk7:clients:createVersionFile
:kafka-0.10.0-jdk7:clients:jar
:kafka-0.10.0-jdk7:core:compileJava UP-TO-DATE
:kafka-0.10.0-jdk7:core:compileScala
:79:
 value DEFAULT_TIMESTAMP in object OffsetCommitRequest is deprecated: see 
corresponding Javadoc for more information.

org.apache.kafka.common.requests.OffsetCommitRequest.DEFAULT_TIMESTAMP
 ^
:36:
 value DEFAULT_TIMESTAMP in object OffsetCommitRequest is deprecated: see 
corresponding Javadoc for more information.
 commitTimestamp: Long = 
org.apache.kafka.common.requests.OffsetCommitRequest.DEFAULT_TIMESTAMP,

  ^
:37:
 value DEFAULT_TIMESTAMP in object OffsetCommitRequest is deprecated: see 
corresponding Javadoc for more information.
 expireTimestamp: Long = 
org.apache.kafka.common.requests.OffsetCommitRequest.DEFAULT_TIMESTAMP) {

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

[Apache Jenkins Server]

See 

Changes:

[ismael] MINOR: Add user overridden test logging events

--
Started by an SCM change
[EnvInject] - Loading node environment variables.
Building remotely on ubuntu-us1 (Ubuntu ubuntu ubuntu-us golang-ppa) in 
workspace 
 > git rev-parse --is-inside-work-tree # timeout=10
Fetching changes from the remote Git repository
 > git config remote.origin.url 
 > https://git-wip-us.apache.org/repos/asf/kafka.git # timeout=10
Fetching upstream changes from https://git-wip-us.apache.org/repos/asf/kafka.git
 > git --version # timeout=10
 > git -c core.askpass=true fetch --tags --progress 
 > https://git-wip-us.apache.org/repos/asf/kafka.git 
 > +refs/heads/*:refs/remotes/origin/*
 > git rev-parse refs/remotes/origin/trunk^{commit} # timeout=10
 > git rev-parse refs/remotes/origin/origin/trunk^{commit} # timeout=10
Checking out Revision 76ca6f8cc1a6533866a2bd7ca6a7d32314d21dba 
(refs/remotes/origin/trunk)
 > git config core.sparsecheckout # timeout=10
 > git checkout -f 76ca6f8cc1a6533866a2bd7ca6a7d32314d21dba
 > git rev-list 7699b91f870f58e71dbdeddba79b58f009f9c0a2 # timeout=10
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK1_8_0_66_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk1.8.0_66
[kafka-trunk-jdk8] $ /bin/bash -xe /tmp/hudson7219891310398566202.sh
+ 
/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2/bin/gradle
To honour the JVM settings for this build a new JVM will be forked. Please 
consider using the daemon: 
http://gradle.org/docs/2.4-rc-2/userguide/gradle_daemon.html.
Building project 'core' with Scala version 2.10.6
:downloadWrapper

BUILD SUCCESSFUL

Total time: 19.483 secs
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK1_8_0_66_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk1.8.0_66
[kafka-trunk-jdk8] $ /bin/bash -xe /tmp/hudson101755821455104654.sh
+ export GRADLE_OPTS=-Xmx1024m
+ GRADLE_OPTS=-Xmx1024m
+ ./gradlew -Dorg.gradle.project.maxParallelForks=1 clean jarAll testAll
To honour the JVM settings for this build a new JVM will be forked. Please 
consider using the daemon: 
https://docs.gradle.org/2.13/userguide/gradle_daemon.html.
Building project 'core' with Scala version 2.10.6
Build file ': 
line 233
useAnt has been deprecated and is scheduled to be removed in Gradle 3.0. The 
Ant-Based Scala compiler is deprecated, please see 
https://docs.gradle.org/current/userguide/scala_plugin.html.
:clean UP-TO-DATE
:clients:clean UP-TO-DATE
:connect:clean UP-TO-DATE
:core:clean
:examples:clean UP-TO-DATE
:log4j-appender:clean UP-TO-DATE
:streams:clean UP-TO-DATE
:tools:clean UP-TO-DATE
:connect:api:clean UP-TO-DATE
:connect:file:clean UP-TO-DATE
:connect:json:clean UP-TO-DATE
:connect:runtime:clean UP-TO-DATE
:streams:examples:clean UP-TO-DATE
:jar_core_2_10
Building project 'core' with Scala version 2.10.6
:kafka-trunk-jdk8:clients:compileJava
:jar_core_2_10 FAILED

FAILURE: Build failed with an exception.

* What went wrong:
Failed to capture snapshot of input files for task 'compileJava' during 
up-to-date check.
> Could not add entry 
> '/home/jenkins/.gradle/caches/modules-2/files-2.1/net.jpountz.lz4/lz4/1.3.0/c708bb2590c0652a642236ef45d9f99ff842a2ce/lz4-1.3.0.jar'
>  to cache fileHashes.bin 
> (

* Try:
Run with --stacktrace option to get the stack trace. Run with --info or --debug 
option to get more log output.

BUILD FAILED

Total time: 15.15 secs
Build step 'Execute shell' marked build as failure
Recording test results
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK1_8_0_66_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk1.8.0_66
ERROR: Step ‘Publish JUnit test result report’ failed: No test report files 
were found. Configuration error?
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK1_8_0_66_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk1.8.0_66

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

[Apache Jenkins Server]

See 

Build failed in Jenkins: kafka-trunk-jdk7 #1336

[Apache Jenkins Server]

See 

Changes:

[ismael] KAFKA-3785; Fetcher spending unnecessary time during metrics recording

[ismael] MINOR: Add user overridden test logging events

--
Started by an SCM change
[EnvInject] - Loading node environment variables.
Building remotely on H11 (docker Ubuntu ubuntu yahoo-not-h2) in workspace 

 > git rev-parse --is-inside-work-tree # timeout=10
Fetching changes from the remote Git repository
 > git config remote.origin.url 
 > https://git-wip-us.apache.org/repos/asf/kafka.git # timeout=10
Fetching upstream changes from https://git-wip-us.apache.org/repos/asf/kafka.git
 > git --version # timeout=10
 > git -c core.askpass=true fetch --tags --progress 
 > https://git-wip-us.apache.org/repos/asf/kafka.git 
 > +refs/heads/*:refs/remotes/origin/*
 > git rev-parse refs/remotes/origin/trunk^{commit} # timeout=10
 > git rev-parse refs/remotes/origin/origin/trunk^{commit} # timeout=10
Checking out Revision 76ca6f8cc1a6533866a2bd7ca6a7d32314d21dba 
(refs/remotes/origin/trunk)
 > git config core.sparsecheckout # timeout=10
 > git checkout -f 76ca6f8cc1a6533866a2bd7ca6a7d32314d21dba
 > git rev-list 1029030466f01937d416e11f93562bcaaecce253 # timeout=10
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK_1_7U51_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk-1.7u51
[kafka-trunk-jdk7] $ /bin/bash -xe /tmp/hudson3952983328994611950.sh
+ 
/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2/bin/gradle
To honour the JVM settings for this build a new JVM will be forked. Please 
consider using the daemon: 
http://gradle.org/docs/2.4-rc-2/userguide/gradle_daemon.html.
Building project 'core' with Scala version 2.10.6
:downloadWrapper

BUILD SUCCESSFUL

Total time: 20.975 secs
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK_1_7U51_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk-1.7u51
[kafka-trunk-jdk7] $ /bin/bash -xe /tmp/hudson629091856357151131.sh
+ export GRADLE_OPTS=-Xmx1024m
+ GRADLE_OPTS=-Xmx1024m
+ ./gradlew -Dorg.gradle.project.maxParallelForks=1 clean jarAll testAll
To honour the JVM settings for this build a new JVM will be forked. Please 
consider using the daemon: 
https://docs.gradle.org/2.13/userguide/gradle_daemon.html.
Building project 'core' with Scala version 2.10.6
Build file ': 
line 233
useAnt has been deprecated and is scheduled to be removed in Gradle 3.0. The 
Ant-Based Scala compiler is deprecated, please see 
https://docs.gradle.org/current/userguide/scala_plugin.html.
:clean UP-TO-DATE
:clients:clean UP-TO-DATE
:connect:clean UP-TO-DATE
:core:clean
:examples:clean UP-TO-DATE
:log4j-appender:clean UP-TO-DATE
:streams:clean UP-TO-DATE
:tools:clean UP-TO-DATE
:connect:api:clean UP-TO-DATE
:connect:file:clean UP-TO-DATE
:connect:json:clean UP-TO-DATE
:connect:runtime:clean UP-TO-DATE
:streams:examples:clean UP-TO-DATE
:jar_core_2_10
Building project 'core' with Scala version 2.10.6
:kafka-trunk-jdk7:clients:compileJava
:jar_core_2_10 FAILED

FAILURE: Build failed with an exception.

* What went wrong:
Failed to capture snapshot of input files for task 'compileJava' during 
up-to-date check.
> Could not add entry 
> '/home/jenkins/.gradle/caches/modules-2/files-2.1/net.jpountz.lz4/lz4/1.3.0/c708bb2590c0652a642236ef45d9f99ff842a2ce/lz4-1.3.0.jar'
>  to cache fileHashes.bin 
> (

* Try:
Run with --stacktrace option to get the stack trace. Run with --info or --debug 
option to get more log output.

BUILD FAILED

Total time: 20.986 secs
Build step 'Execute shell' marked build as failure
Recording test results
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK_1_7U51_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk-1.7u51
ERROR: Step ?Publish JUnit test result report? failed: No test report files 
were found. Configuration error?
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK_1_7U51_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk-1.7u51

[GitHub] kafka pull request #1456: MINOR: Add user overridden test logging events

[asfgit]

Github user asfgit closed the pull request at:

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


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Updated] (KAFKA-3787) Preserve message timestamp in mirror mkaer

[Ismael Juma (JIRA)]

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

Ismael Juma updated KAFKA-3787:
---
Fix Version/s: 0.10.0.1

> Preserve message timestamp in mirror mkaer
> --
>
> Key: KAFKA-3787
> URL: https://issues.apache.org/jira/browse/KAFKA-3787
> Project: Kafka
>  Issue Type: Bug
>  Components: tools
>Affects Versions: 0.10.0.0
>Reporter: TAO XIAO
>Assignee: TAO XIAO
> Fix For: 0.10.0.1
>
>
> The timestamp of messages consumed by mirror maker is not preserved after 
> sending to target cluster. The correct behavior is to keep create timestamp 
> the same in both source and target clusters.
> Here is the KIP-32 that illustrates the correct behavior
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-32+-+Add+timestamps+to+Kafka+message#KIP-32-AddtimestampstoKafkamessage-Mirrormakercaseindetail



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Build failed in Jenkins: kafka-0.10.0-jdk7 #113

[Apache Jenkins Server]

See 

Changes:

[ismael] KAFKA-3785; Fetcher spending unnecessary time during metrics recording

--
Started by an SCM change
[EnvInject] - Loading node environment variables.
Building remotely on H10 (docker Ubuntu ubuntu yahoo-not-h2) in workspace 

 > git rev-parse --is-inside-work-tree # timeout=10
Fetching changes from the remote Git repository
 > git config remote.origin.url 
 > https://git-wip-us.apache.org/repos/asf/kafka.git # timeout=10
Fetching upstream changes from https://git-wip-us.apache.org/repos/asf/kafka.git
 > git --version # timeout=10
 > git -c core.askpass=true fetch --tags --progress 
 > https://git-wip-us.apache.org/repos/asf/kafka.git 
 > +refs/heads/*:refs/remotes/origin/*
 > git rev-parse refs/remotes/origin/0.10.0^{commit} # timeout=10
 > git rev-parse refs/remotes/origin/origin/0.10.0^{commit} # timeout=10
Checking out Revision c91f742deb8b83d59847f2afb7a786f18b34eb1d 
(refs/remotes/origin/0.10.0)
 > git config core.sparsecheckout # timeout=10
 > git checkout -f c91f742deb8b83d59847f2afb7a786f18b34eb1d
 > git rev-list de23c6376bead90249e1e4344d8b7a5ed148fba3 # timeout=10
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK_1_7U51_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk-1.7u51
[kafka-0.10.0-jdk7] $ /bin/bash -xe /tmp/hudson7822978477387100098.sh
+ 
/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2/bin/gradle
To honour the JVM settings for this build a new JVM will be forked. Please 
consider using the daemon: 
http://gradle.org/docs/2.4-rc-2/userguide/gradle_daemon.html.
Building project 'core' with Scala version 2.10.6
:downloadWrapper

BUILD SUCCESSFUL

Total time: 20.935 secs
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK_1_7U51_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk-1.7u51
[kafka-0.10.0-jdk7] $ /bin/bash -xe /tmp/hudson3472525615863298250.sh
+ export GRADLE_OPTS=-Xmx1024m
+ GRADLE_OPTS=-Xmx1024m
+ ./gradlew -Dorg.gradle.project.maxParallelForks=1 clean jarAll testAll
To honour the JVM settings for this build a new JVM will be forked. Please 
consider using the daemon: 
https://docs.gradle.org/2.13/userguide/gradle_daemon.html.
Building project 'core' with Scala version 2.10.6
Build file ': 
line 231
useAnt has been deprecated and is scheduled to be removed in Gradle 3.0. The 
Ant-Based Scala compiler is deprecated, please see 
https://docs.gradle.org/current/userguide/scala_plugin.html.
:clean UP-TO-DATE
:clients:clean UP-TO-DATE
:connect:clean UP-TO-DATE
:core:clean
:examples:clean UP-TO-DATE
:log4j-appender:clean UP-TO-DATE
:streams:clean UP-TO-DATE
:tools:clean UP-TO-DATE
:connect:api:clean UP-TO-DATE
:connect:file:clean UP-TO-DATE
:connect:json:clean UP-TO-DATE
:connect:runtime:clean UP-TO-DATE
:streams:examples:clean UP-TO-DATE
:jar_core_2_10
Building project 'core' with Scala version 2.10.6
:kafka-0.10.0-jdk7:clients:compileJava
:jar_core_2_10 FAILED

FAILURE: Build failed with an exception.

* What went wrong:
org.gradle.api.internal.changedetection.state.FileCollectionSnapshotImpl cannot 
be cast to 
org.gradle.api.internal.changedetection.state.OutputFilesCollectionSnapshotter$OutputFilesSnapshot

* Try:
Run with --stacktrace option to get the stack trace. Run with --info or --debug 
option to get more log output.

BUILD FAILED

Total time: 21.352 secs
Build step 'Execute shell' marked build as failure
Recording test results
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK_1_7U51_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk-1.7u51
ERROR: Step ?Publish JUnit test result report? failed: No test report files 
were found. Configuration error?
Setting 
GRADLE_2_4_RC_2_HOME=/home/jenkins/jenkins-slave/tools/hudson.plugins.gradle.GradleInstallation/Gradle_2.4-rc-2
Setting 
JDK_1_7U51_HOME=/home/jenkins/jenkins-slave/tools/hudson.model.JDK/jdk-1.7u51

[jira] [Updated] (KAFKA-3787) Preserve message timestamp in mirror mkaer

[TAO XIAO (JIRA)]

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

TAO XIAO updated KAFKA-3787:

Status: Patch Available  (was: Open)

> Preserve message timestamp in mirror mkaer
> --
>
> Key: KAFKA-3787
> URL: https://issues.apache.org/jira/browse/KAFKA-3787
> Project: Kafka
>  Issue Type: Bug
>  Components: tools
>Affects Versions: 0.10.0.0
>Reporter: TAO XIAO
>Assignee: TAO XIAO
>
> The timestamp of messages consumed by mirror maker is not preserved after 
> sending to target cluster. The correct behavior is to keep create timestamp 
> the same in both source and target clusters.
> Here is the KIP-32 that illustrates the correct behavior
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-32+-+Add+timestamps+to+Kafka+message#KIP-32-AddtimestampstoKafkamessage-Mirrormakercaseindetail



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-3787) Preserve message timestamp in mirror mkaer

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-3787?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15313886#comment-15313886
 ] 

ASF GitHub Bot commented on KAFKA-3787:
---

GitHub user xiaotao183 opened a pull request:

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

KAFKA-3787: Preserve the message timestamp in mirror maker

The timestamp of messages consumed by mirror maker is not preserved after 
sending to target cluster. The correct behavior is to keep create timestamp the 
same in both source and target clusters.

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/xiaotao183/kafka KAFKA-3787

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/1466.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #1466


commit 29b874b01bfdd1d7114e5eb431307b5031a80754
Author: Tao Xiao 
Date:   2016-06-03T09:08:37Z

KAFKA-3787: Preserve the message timestamp in mirror maker




> Preserve message timestamp in mirror mkaer
> --
>
> Key: KAFKA-3787
> URL: https://issues.apache.org/jira/browse/KAFKA-3787
> Project: Kafka
>  Issue Type: Bug
>  Components: tools
>Affects Versions: 0.10.0.0
>Reporter: TAO XIAO
>Assignee: TAO XIAO
>
> The timestamp of messages consumed by mirror maker is not preserved after 
> sending to target cluster. The correct behavior is to keep create timestamp 
> the same in both source and target clusters.
> Here is the KIP-32 that illustrates the correct behavior
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-32+-+Add+timestamps+to+Kafka+message#KIP-32-AddtimestampstoKafkamessage-Mirrormakercaseindetail



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] kafka pull request #1466: KAFKA-3787: Preserve the message timestamp in mirr...

[xiaotao183]

GitHub user xiaotao183 opened a pull request:

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

KAFKA-3787: Preserve the message timestamp in mirror maker

The timestamp of messages consumed by mirror maker is not preserved after 
sending to target cluster. The correct behavior is to keep create timestamp the 
same in both source and target clusters.

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/xiaotao183/kafka KAFKA-3787

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/1466.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #1466


commit 29b874b01bfdd1d7114e5eb431307b5031a80754
Author: Tao Xiao 
Date:   2016-06-03T09:08:37Z

KAFKA-3787: Preserve the message timestamp in mirror maker




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Created] (KAFKA-3787) Preserve message timestamp in mirror mkaer

[TAO XIAO (JIRA)]

TAO XIAO created KAFKA-3787:
---

 Summary: Preserve message timestamp in mirror mkaer
 Key: KAFKA-3787
 URL: https://issues.apache.org/jira/browse/KAFKA-3787
 Project: Kafka
  Issue Type: Bug
  Components: tools
Affects Versions: 0.10.0.0
Reporter: TAO XIAO
Assignee: TAO XIAO


The timestamp of messages consumed by mirror maker is not preserved after 
sending to target cluster. The correct behavior is to keep create timestamp the 
same in both source and target clusters.

Here is the KIP-32 that illustrates the correct behavior
https://cwiki.apache.org/confluence/display/KAFKA/KIP-32+-+Add+timestamps+to+Kafka+message#KIP-32-AddtimestampstoKafkamessage-Mirrormakercaseindetail



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-3785) Fetcher spending unnecessary time during metrics recording

[Ismael Juma (JIRA)]

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

Ismael Juma updated KAFKA-3785:
---
Assignee: Greg Fodor

> Fetcher spending unnecessary time during metrics recording
> --
>
> Key: KAFKA-3785
> URL: https://issues.apache.org/jira/browse/KAFKA-3785
> Project: Kafka
>  Issue Type: Improvement
>  Components: consumer
>Reporter: Greg Fodor
>Assignee: Greg Fodor
> Fix For: 0.10.0.1
>
>
> Profiling a Kafka Streams job revealed some hotspots in the Fetcher during 
> metrics flushing. Previous discussion here:
> https://issues.apache.org/jira/browse/KAFKA-3769



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-3785) Fetcher spending unnecessary time during metrics recording

[Ismael Juma (JIRA)]

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

Ismael Juma updated KAFKA-3785:
---
Fix Version/s: 0.10.0.1

> Fetcher spending unnecessary time during metrics recording
> --
>
> Key: KAFKA-3785
> URL: https://issues.apache.org/jira/browse/KAFKA-3785
> Project: Kafka
>  Issue Type: Improvement
>  Components: consumer
>Reporter: Greg Fodor
> Fix For: 0.10.0.1
>
>
> Profiling a Kafka Streams job revealed some hotspots in the Fetcher during 
> metrics flushing. Previous discussion here:
> https://issues.apache.org/jira/browse/KAFKA-3769



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Resolved] (KAFKA-3785) Fetcher spending unnecessary time during metrics recording

[Ismael Juma (JIRA)]

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

Ismael Juma resolved KAFKA-3785.

Resolution: Fixed

> Fetcher spending unnecessary time during metrics recording
> --
>
> Key: KAFKA-3785
> URL: https://issues.apache.org/jira/browse/KAFKA-3785
> Project: Kafka
>  Issue Type: Improvement
>  Components: consumer
>Reporter: Greg Fodor
> Fix For: 0.10.0.1
>
>
> Profiling a Kafka Streams job revealed some hotspots in the Fetcher during 
> metrics flushing. Previous discussion here:
> https://issues.apache.org/jira/browse/KAFKA-3769



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-3785) Fetcher spending unnecessary time during metrics recording

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-3785?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15313862#comment-15313862
 ] 

ASF GitHub Bot commented on KAFKA-3785:
---

Github user asfgit closed the pull request at:

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


> Fetcher spending unnecessary time during metrics recording
> --
>
> Key: KAFKA-3785
> URL: https://issues.apache.org/jira/browse/KAFKA-3785
> Project: Kafka
>  Issue Type: Improvement
>  Components: consumer
>Reporter: Greg Fodor
>
> Profiling a Kafka Streams job revealed some hotspots in the Fetcher during 
> metrics flushing. Previous discussion here:
> https://issues.apache.org/jira/browse/KAFKA-3769



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] kafka pull request #1464: KAFKA-3785; Fetcher spending unnecessary time duri...

[asfgit]

Github user asfgit closed the pull request at:

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


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Comment Edited] (KAFKA-3775) Throttle maximum number of tasks assigned to a single KafkaStreams

[Yuto Kawamura (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-3775?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15313828#comment-15313828
 ] 

Yuto Kawamura edited comment on KAFKA-3775 at 6/3/16 8:17 AM:
--

Thanks for feedback [~BigAndy] .

> With the purposed design some partitions would remain with out a consumer. 
> This seems like a fundamental switch away from Kafka's current model, and a 
> risky one in IMHO.

Some partitions would remain without a consumer *if the number of living 
instances become lower than the number of {{num of partitions / 
max.tasks.assigned}}*.
Let's say you have 100 partitions and launching 50 KafkaStreams instances with 
setting {{max.tasks.assigned=5}}. When you started all 50 instances each 
instance might get 2 partitions assigned, which is the desired distribution.
Then what will happen when an instance failed? 2 partitions which were held by 
the dead instance will be reassigned to remaining instances without any problem 
as other instances still have plenty number of {{max.tasks.assigned}}.
If more than 31 instances dead at the moment, yes, some partitions will be 
remain unassigned, but this is out of consideration as the value of 
{{max.tasks.assigned}} was determined with the consideration to the amount of 
system resources(CPU, mem, network bandwidth), which means these unassigned 
partitions could never be processed normally even they reassigned to the living 
instances because of hardware resource is limited.

> This seems like a fundamental switch away from Kafka's current model, and a 
> risky one in IMHO.

BTW, may I ask what you meant by "Kafka's current model" and what risk could 
you expect much concretely?(user won't noticed unassigned partitions existence?)

> Could you also elaborate on why settings such as 'max.poll.records' don't 
> help stop your initial instance going pop? Maybe there are other alternative 
> solutions here...

Because even I set {{max.poll.records}} to lower, it reduced the number of 
records fetched by single Fetch request but instead the number of Fetch request 
will be increased. That means the total throughput wouldn't chagne which still 
leads traffic bursting.
At the same time, it doesn't make sense to me that adjusting the value of 
{{max.poll.records}} with expecting that a single instance gets all partitions 
assigned, as I can set that value to much higher practically when other 
instances join the group and partitions are evenly distributed.



was (Author: kawamuray):
Thanks for feedback [~BigAndy] .

> With the purposed design some partitions would remain with out a consumer. 
> This seems like a fundamental switch away from Kafka's current model, and a 
> risky one in IMHO.

Some partitions would remain without a consumer *if the number of living 
instances become lower than the number of {{num of partitions / 
max.tasks.assigned}}*.
Let's say you have 100 partitions and launching 50 KafkaStreams instances with 
setting {{max.tasks.assigned=5}}. When you started all 50 instances each 
instance might get 2 partitions assigned, which is the desired distribution.
Then what will happen when an instance failed? 2 partitions which were held by 
the dead instance will be reassigned to remaining instances without any problem 
as other instances still have plenty number of {{max.tasks.assigned}}.
If more than 31 instances dead at the moment, yes, some partitions will be 
remain unassigned, but this is out of consideration as the value of 
{{max.tasks.assigned}} was determined with the consideration to the amount of 
system resources(CPU, mem, network bandwidth), which means these unassigned 
partitions could never be processed normally even they reassigned to the living 
instances because of hardware resource is limited.

> This seems like a fundamental switch away from Kafka's current model, and a 
> risky one in IMHO.

BTW, may I ask what you meant by "Kafka's current model" and what risk could 
you expect much concretely?(user won't noticed unassigned partitions existence?)

> Could you also elaborate on why settings such as 'max.poll.records' don't 
> help stop your initial instance going pop? Maybe there are other alternative 
> solutions here...

Because even I set {{max.poll.records}} to lower, it reduced the number of 
records fetched by single Fetch request but instead the number of Fetch request 
will be increased. That means the total throughput wouldn't chagne which still 
leads traffic bursting.
At the same time, it doesn't make sense to me that adjusting the value of 
{{max.poll.records}} with expecting that a single gets all partitions assigned, 
as I can set that value to much higher practically when other instances join 
the group and partitions are evenly distributed.


> Throttle maximum number of tasks assigned to a single KafkaStreams
> --
>
> Key: 

[jira] [Commented] (KAFKA-3775) Throttle maximum number of tasks assigned to a single KafkaStreams

[Yuto Kawamura (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-3775?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15313828#comment-15313828
 ] 

Yuto Kawamura commented on KAFKA-3775:
--

Thanks for feedback [~BigAndy] .

> With the purposed design some partitions would remain with out a consumer. 
> This seems like a fundamental switch away from Kafka's current model, and a 
> risky one in IMHO.

Some partitions would remain without a consumer *if the number of living 
instances become lower than the number of {{num of partitions / 
max.tasks.assigned}}*.
Let's say you have 100 partitions and launching 50 KafkaStreams instances with 
setting {{max.tasks.assigned=5}}. When you started all 50 instances each 
instance might get 2 partitions assigned, which is the desired distribution.
Then what will happen when an instance failed? 2 partitions which were held by 
the dead instance will be reassigned to remaining instances without any problem 
as other instances still have plenty number of {{max.tasks.assigned}}.
If more than 31 instances dead at the moment, yes, some partitions will be 
remain unassigned, but this is out of consideration as the value of 
{{max.tasks.assigned}} was determined with the consideration to the amount of 
system resources(CPU, mem, network bandwidth), which means these unassigned 
partitions could never be processed normally even they reassigned to the living 
instances because of hardware resource is limited.

> This seems like a fundamental switch away from Kafka's current model, and a 
> risky one in IMHO.

BTW, may I ask what you meant by "Kafka's current model" and what risk could 
you expect much concretely?(user won't noticed unassigned partitions existence?)

> Could you also elaborate on why settings such as 'max.poll.records' don't 
> help stop your initial instance going pop? Maybe there are other alternative 
> solutions here...

Because even I set {{max.poll.records}} to lower, it reduced the number of 
records fetched by single Fetch request but instead the number of Fetch request 
will be increased. That means the total throughput wouldn't chagne which still 
leads traffic bursting.
At the same time, it doesn't make sense to me that adjusting the value of 
{{max.poll.records}} with expecting that a single gets all partitions assigned, 
as I can set that value to much higher practically when other instances join 
the group and partitions are evenly distributed.


> Throttle maximum number of tasks assigned to a single KafkaStreams
> --
>
> Key: KAFKA-3775
> URL: https://issues.apache.org/jira/browse/KAFKA-3775
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Affects Versions: 0.10.0.0
>Reporter: Yuto Kawamura
>Assignee: Yuto Kawamura
> Fix For: 0.10.1.0
>
>
> As of today, if I start a Kafka Streams app on a single machine which 
> consists of single KafkaStreams instance, that instance gets all partitions 
> of the target topic assigned.
> As we're using it to process topics which has huge number of partitions and 
> message traffic, it is a problem that we don't have a way of throttling the 
> maximum amount of partitions assigned to a single instance.
> In fact, when we started a Kafka Streams app which consumes a topic which has 
> more than 10MB/sec traffic of each partition we saw that all partitions 
> assigned to the first instance and soon the app dead by OOM.
> I know that there's some workarounds considerable here. for example:
> - Start multiple instances at once so the partitions distributed evenly.
>   => Maybe works. but as Kafka Streams is a library but not an execution 
> framework, there's no predefined procedure of starting Kafka Streams apps so 
> some users might wanna take an option to start the first single instance and 
> check if it works as expected with lesster number of partitions(I want :p)
> - Adjust config parameters such as {{buffered.records.per.partition}}, 
> {{max.partition.fetch.bytes}} and {{max.poll.records}} to reduce the heap 
> pressure.
>   => Maybe works. but still have two problems IMO:
>   - Still leads traffic explosion with high throughput processing as it 
> accepts all incoming messages from hundreads of partitions.
>   - In the first place, by the distributed system principle, it's wired that 
> users don't have a away to control maximum "partitions" assigned to a single 
> shard(an instance of KafkaStreams here). Users should be allowed to provide 
> the maximum amount of partitions that is considered as possible to be 
> processed with single instance(or host).
> Here, I'd like to introduce a new configuration parameter 
> {{max.tasks.assigned}}, which limits the number of tasks(a notion of 
> partition) assigned to the processId(which is the notion of single 
> KafkaStreams instance).
> 

[jira] [Comment Edited] (KAFKA-3775) Throttle maximum number of tasks assigned to a single KafkaStreams

[The Data Lorax (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-3775?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15313773#comment-15313773
 ] 

The Data Lorax edited comment on KAFKA-3775 at 6/3/16 7:24 AM:
---

I wonder how such a design would maintain the level of resiliency currently 
offered? At the moment, in a running multi-process cluster, the other processes 
pick up the slack if one of them should fail. With the purposed design some 
partitions would remain with out a consumer. This seems like a fundamental 
switch away from Kafka's current model, and a risky one in IMHO.

Could you also elaborate on why settings such as 'max.poll.records' don't help 
stop your initial instance going pop? Maybe there are other alternative 
solutions here... 


was (Author: bigandy):
I wonder how surah a design would maintain the level of resiliency currently 
offered? At the moment, in a running multi-process cluster the other processes 
pick up the slack if one of them should fail. With the purposed design some 
partitions would remain with out a consumer. This seems like a fundamental 
switch away from Kafka's current model.

Could you also elaborate on why settings such as 'max.poll.records' don't help 
stop your initial instance going pop? Maybe there are other alternative 
solutions here... 

> Throttle maximum number of tasks assigned to a single KafkaStreams
> --
>
> Key: KAFKA-3775
> URL: https://issues.apache.org/jira/browse/KAFKA-3775
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Affects Versions: 0.10.0.0
>Reporter: Yuto Kawamura
>Assignee: Yuto Kawamura
> Fix For: 0.10.1.0
>
>
> As of today, if I start a Kafka Streams app on a single machine which 
> consists of single KafkaStreams instance, that instance gets all partitions 
> of the target topic assigned.
> As we're using it to process topics which has huge number of partitions and 
> message traffic, it is a problem that we don't have a way of throttling the 
> maximum amount of partitions assigned to a single instance.
> In fact, when we started a Kafka Streams app which consumes a topic which has 
> more than 10MB/sec traffic of each partition we saw that all partitions 
> assigned to the first instance and soon the app dead by OOM.
> I know that there's some workarounds considerable here. for example:
> - Start multiple instances at once so the partitions distributed evenly.
>   => Maybe works. but as Kafka Streams is a library but not an execution 
> framework, there's no predefined procedure of starting Kafka Streams apps so 
> some users might wanna take an option to start the first single instance and 
> check if it works as expected with lesster number of partitions(I want :p)
> - Adjust config parameters such as {{buffered.records.per.partition}}, 
> {{max.partition.fetch.bytes}} and {{max.poll.records}} to reduce the heap 
> pressure.
>   => Maybe works. but still have two problems IMO:
>   - Still leads traffic explosion with high throughput processing as it 
> accepts all incoming messages from hundreads of partitions.
>   - In the first place, by the distributed system principle, it's wired that 
> users don't have a away to control maximum "partitions" assigned to a single 
> shard(an instance of KafkaStreams here). Users should be allowed to provide 
> the maximum amount of partitions that is considered as possible to be 
> processed with single instance(or host).
> Here, I'd like to introduce a new configuration parameter 
> {{max.tasks.assigned}}, which limits the number of tasks(a notion of 
> partition) assigned to the processId(which is the notion of single 
> KafkaStreams instance).
> At the same time we need to change StreamPartitionAssignor(TaskAssignor) to 
> tolerate the incomplete assignment. That is, Kafka Streams should continue 
> working for the part of partitions even there are some partitions left 
> unassigned, in order to satisfy this> "user may want to take an option to 
> start the first single instance and check if it works as expected with 
> lesster number of partitions(I want :p)".
> I've implemented the rough POC for this. PTAL and if it make sense I will 
> continue sophisticating it.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-3775) Throttle maximum number of tasks assigned to a single KafkaStreams

[The Data Lorax (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-3775?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15313773#comment-15313773
 ] 

The Data Lorax commented on KAFKA-3775:
---

I wonder how surah a design would maintain the level of resiliency currently 
offered? At the moment, in a running multi-process cluster the other processes 
pick up the slack if one of them should fail. With the purposed design some 
partitions would remain with out a consumer. This seems like a fundamental 
switch away from Kafka's current model.

Could you also elaborate on why settings such as 'max.poll.records' don't help 
stop your initial instance going pop? Maybe there are other alternative 
solutions here... 

> Throttle maximum number of tasks assigned to a single KafkaStreams
> --
>
> Key: KAFKA-3775
> URL: https://issues.apache.org/jira/browse/KAFKA-3775
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Affects Versions: 0.10.0.0
>Reporter: Yuto Kawamura
>Assignee: Yuto Kawamura
> Fix For: 0.10.1.0
>
>
> As of today, if I start a Kafka Streams app on a single machine which 
> consists of single KafkaStreams instance, that instance gets all partitions 
> of the target topic assigned.
> As we're using it to process topics which has huge number of partitions and 
> message traffic, it is a problem that we don't have a way of throttling the 
> maximum amount of partitions assigned to a single instance.
> In fact, when we started a Kafka Streams app which consumes a topic which has 
> more than 10MB/sec traffic of each partition we saw that all partitions 
> assigned to the first instance and soon the app dead by OOM.
> I know that there's some workarounds considerable here. for example:
> - Start multiple instances at once so the partitions distributed evenly.
>   => Maybe works. but as Kafka Streams is a library but not an execution 
> framework, there's no predefined procedure of starting Kafka Streams apps so 
> some users might wanna take an option to start the first single instance and 
> check if it works as expected with lesster number of partitions(I want :p)
> - Adjust config parameters such as {{buffered.records.per.partition}}, 
> {{max.partition.fetch.bytes}} and {{max.poll.records}} to reduce the heap 
> pressure.
>   => Maybe works. but still have two problems IMO:
>   - Still leads traffic explosion with high throughput processing as it 
> accepts all incoming messages from hundreads of partitions.
>   - In the first place, by the distributed system principle, it's wired that 
> users don't have a away to control maximum "partitions" assigned to a single 
> shard(an instance of KafkaStreams here). Users should be allowed to provide 
> the maximum amount of partitions that is considered as possible to be 
> processed with single instance(or host).
> Here, I'd like to introduce a new configuration parameter 
> {{max.tasks.assigned}}, which limits the number of tasks(a notion of 
> partition) assigned to the processId(which is the notion of single 
> KafkaStreams instance).
> At the same time we need to change StreamPartitionAssignor(TaskAssignor) to 
> tolerate the incomplete assignment. That is, Kafka Streams should continue 
> working for the part of partitions even there are some partitions left 
> unassigned, in order to satisfy this> "user may want to take an option to 
> start the first single instance and check if it works as expected with 
> lesster number of partitions(I want :p)".
> I've implemented the rough POC for this. PTAL and if it make sense I will 
> continue sophisticating it.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-3775) Throttle maximum number of tasks assigned to a single KafkaStreams

[Guozhang Wang (JIRA)]

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

Guozhang Wang updated KAFKA-3775:
-
Reviewer: Matthias J. Sax

> Throttle maximum number of tasks assigned to a single KafkaStreams
> --
>
> Key: KAFKA-3775
> URL: https://issues.apache.org/jira/browse/KAFKA-3775
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Affects Versions: 0.10.0.0
>Reporter: Yuto Kawamura
>Assignee: Yuto Kawamura
> Fix For: 0.10.1.0
>
>
> As of today, if I start a Kafka Streams app on a single machine which 
> consists of single KafkaStreams instance, that instance gets all partitions 
> of the target topic assigned.
> As we're using it to process topics which has huge number of partitions and 
> message traffic, it is a problem that we don't have a way of throttling the 
> maximum amount of partitions assigned to a single instance.
> In fact, when we started a Kafka Streams app which consumes a topic which has 
> more than 10MB/sec traffic of each partition we saw that all partitions 
> assigned to the first instance and soon the app dead by OOM.
> I know that there's some workarounds considerable here. for example:
> - Start multiple instances at once so the partitions distributed evenly.
>   => Maybe works. but as Kafka Streams is a library but not an execution 
> framework, there's no predefined procedure of starting Kafka Streams apps so 
> some users might wanna take an option to start the first single instance and 
> check if it works as expected with lesster number of partitions(I want :p)
> - Adjust config parameters such as {{buffered.records.per.partition}}, 
> {{max.partition.fetch.bytes}} and {{max.poll.records}} to reduce the heap 
> pressure.
>   => Maybe works. but still have two problems IMO:
>   - Still leads traffic explosion with high throughput processing as it 
> accepts all incoming messages from hundreads of partitions.
>   - In the first place, by the distributed system principle, it's wired that 
> users don't have a away to control maximum "partitions" assigned to a single 
> shard(an instance of KafkaStreams here). Users should be allowed to provide 
> the maximum amount of partitions that is considered as possible to be 
> processed with single instance(or host).
> Here, I'd like to introduce a new configuration parameter 
> {{max.tasks.assigned}}, which limits the number of tasks(a notion of 
> partition) assigned to the processId(which is the notion of single 
> KafkaStreams instance).
> At the same time we need to change StreamPartitionAssignor(TaskAssignor) to 
> tolerate the incomplete assignment. That is, Kafka Streams should continue 
> working for the part of partitions even there are some partitions left 
> unassigned, in order to satisfy this> "user may want to take an option to 
> start the first single instance and check if it works as expected with 
> lesster number of partitions(I want :p)".
> I've implemented the rough POC for this. PTAL and if it make sense I will 
> continue sophisticating it.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)