You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@kafka.apache.org by Jun Rao <ju...@gmail.com> on 2015/05/01 02:01:31 UTC
Re: Review Request 33548: KAFKA-2068 Step I: merge in KAFKA-1841
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/33548/#review82224
-----------------------------------------------------------
Ship it!
Thanks for the patch. +1. Just a few minor comments below. Also, could you trying using the following test code in 0.8.1 to see if v0 OffsetCommitRequest still works with 0.8.3 broker?
package kafka
import kafka.utils.Logging
import kafka.consumer.SimpleConsumer
import kafka.common.{OffsetMetadataAndError, TopicAndPartition}
import kafka.api.{OffsetFetchRequest, OffsetCommitRequest}
object OffsetCommitMain extends Logging {
def main(args: Array[String]): Unit = {
val simpleConsumer = new SimpleConsumer("localhost", 9092, 1000000, 64*1024, "test-client")
val topic = "topic"
// Commit an offset
val topicAndPartition = TopicAndPartition(topic, 0)
val commitRequest = OffsetCommitRequest("test-group", Map(topicAndPartition -> OffsetMetadataAndError(offset=42L)))
val commitResponse = simpleConsumer.commitOffsets(commitRequest)
System.out.println("OffsetCommitResponse: " + commitResponse.toString())
// Fetch it and verify
val fetchRequest = OffsetFetchRequest("test-group", Seq(topicAndPartition))
val fetchResponse = simpleConsumer.fetchOffsets(fetchRequest)
System.out.println("OffsetFetchResponse: " + fetchResponse.toString())
}
}
clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java
<https://reviews.apache.org/r/33548/#comment132925>
It's probably clearer if we define OFFSET_FETCH_RESPONSE_V1.
core/src/test/scala/unit/kafka/server/OffsetCommitTest.scala
<https://reviews.apache.org/r/33548/#comment132927>
should not exist => Committed offset should not exist
- Jun Rao
On April 25, 2015, 7:59 a.m., Guozhang Wang wrote:
>
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/33548/
> -----------------------------------------------------------
>
> (Updated April 25, 2015, 7:59 a.m.)
>
>
> Review request for kafka.
>
>
> Bugs: KAFKA-2068
> https://issues.apache.org/jira/browse/KAFKA-2068
>
>
> Repository: kafka
>
>
> Description
> -------
>
> 1. Remove timestamp in partition-info for offset-commit-request.v0
> 2. Handle offset-commit-request.v0 by writting to ZK.
> 3. Add offset-fetch-request.v1 with the same format as offset-fetch-request.v0, which expects the same version of offset-fetch-response (v0).
> 4. Handle offset-fetch-request.v0 by reading from ZK.
> 5. Minor changes in unit tests
>
>
> Diffs
> -----
>
> clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java d53fe45b9c5d5c873facd9696b1eacb67e812bca
> clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java a0e19768ff400d74c87b592f6c25c666696727d2
> core/src/main/scala/kafka/api/OffsetCommitRequest.scala cf8e6acc426aef6eb19d862bf6a108a5fc37907a
> core/src/main/scala/kafka/api/OffsetFetchRequest.scala 67811a752a470bf9bdbc8c5419e8d6e20a006169
> core/src/main/scala/kafka/common/OffsetMetadataAndError.scala 139913f2a40a9afdf3baa7044af265afdebc1fda
> core/src/main/scala/kafka/server/KafkaApis.scala b4004aa3a1456d337199aa1245fb0ae61f6add46
> core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala dbf9f48fac0150bc2f1e655030c67c21bd160735
> core/src/test/scala/unit/kafka/server/OffsetCommitTest.scala 652208a70f66045b854549d93cbbc2b77c24b10b
>
> Diff: https://reviews.apache.org/r/33548/diff/
>
>
> Testing
> -------
>
> Unit tests.
>
>
> Thanks,
>
> Guozhang Wang
>
>
Re: Review Request 33548: KAFKA-2068 Step I: merge in KAFKA-1841
Posted by Guozhang Wang <wa...@gmail.com>.
> On May 1, 2015, 12:01 a.m., Jun Rao wrote:
> > Thanks for the patch. +1. Just a few minor comments below. Also, could you trying using the following test code in 0.8.1 to see if v0 OffsetCommitRequest still works with 0.8.3 broker?
> >
> >
> > package kafka
> >
> > import kafka.utils.Logging
> > import kafka.consumer.SimpleConsumer
> > import kafka.common.{OffsetMetadataAndError, TopicAndPartition}
> > import kafka.api.{OffsetFetchRequest, OffsetCommitRequest}
> >
> >
> > object OffsetCommitMain extends Logging {
> >
> > def main(args: Array[String]): Unit = {
> > val simpleConsumer = new SimpleConsumer("localhost", 9092, 1000000, 64*1024, "test-client")
> >
> > val topic = "topic"
> > // Commit an offset
> > val topicAndPartition = TopicAndPartition(topic, 0)
> > val commitRequest = OffsetCommitRequest("test-group", Map(topicAndPartition -> OffsetMetadataAndError(offset=42L)))
> > val commitResponse = simpleConsumer.commitOffsets(commitRequest)
> >
> > System.out.println("OffsetCommitResponse: " + commitResponse.toString())
> >
> > // Fetch it and verify
> > val fetchRequest = OffsetFetchRequest("test-group", Seq(topicAndPartition))
> > val fetchResponse = simpleConsumer.fetchOffsets(fetchRequest)
> >
> > System.out.println("OffsetFetchResponse: " + fetchResponse.toString())
> >
> >
> > }
> > }
Verified using the test code.
- Guozhang
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/33548/#review82224
-----------------------------------------------------------
On April 25, 2015, 7:59 a.m., Guozhang Wang wrote:
>
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/33548/
> -----------------------------------------------------------
>
> (Updated April 25, 2015, 7:59 a.m.)
>
>
> Review request for kafka.
>
>
> Bugs: KAFKA-2068
> https://issues.apache.org/jira/browse/KAFKA-2068
>
>
> Repository: kafka
>
>
> Description
> -------
>
> 1. Remove timestamp in partition-info for offset-commit-request.v0
> 2. Handle offset-commit-request.v0 by writting to ZK.
> 3. Add offset-fetch-request.v1 with the same format as offset-fetch-request.v0, which expects the same version of offset-fetch-response (v0).
> 4. Handle offset-fetch-request.v0 by reading from ZK.
> 5. Minor changes in unit tests
>
>
> Diffs
> -----
>
> clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java d53fe45b9c5d5c873facd9696b1eacb67e812bca
> clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java a0e19768ff400d74c87b592f6c25c666696727d2
> core/src/main/scala/kafka/api/OffsetCommitRequest.scala cf8e6acc426aef6eb19d862bf6a108a5fc37907a
> core/src/main/scala/kafka/api/OffsetFetchRequest.scala 67811a752a470bf9bdbc8c5419e8d6e20a006169
> core/src/main/scala/kafka/common/OffsetMetadataAndError.scala 139913f2a40a9afdf3baa7044af265afdebc1fda
> core/src/main/scala/kafka/server/KafkaApis.scala b4004aa3a1456d337199aa1245fb0ae61f6add46
> core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala dbf9f48fac0150bc2f1e655030c67c21bd160735
> core/src/test/scala/unit/kafka/server/OffsetCommitTest.scala 652208a70f66045b854549d93cbbc2b77c24b10b
>
> Diff: https://reviews.apache.org/r/33548/diff/
>
>
> Testing
> -------
>
> Unit tests.
>
>
> Thanks,
>
> Guozhang Wang
>
>