Re: Kafka svn repo removed from incubator
I see someone filed a jira, we'll answer there Neha Narkhede wrote on Wed, Nov 28, 2012 at 16:43:08 -0800: Gavin, Thanks for moving the repository. Any idea what needs to be done to get the website moved ? -Neha On Wed, Nov 28, 2012 at 4:08 PM, Gavin McDonald ga...@16degrees.com.au wrote: Yep http://svn.apache.org/repos/asf/kafka/ Your project was guinea pig for a new incubator to tlp workflow. Kafkas move to TLP is complete , svn, mailing lists all done. We still need to move your website but that's about it I think. Apologies that you were not notified in advance. Gav. From: Jun Rao [mailto:jun...@gmail.com] Sent: Thursday, 29 November 2012 4:57 AM To: Apache Infrastructure Cc: kafka-...@incubator.apache.org; kafka-us...@incubator.apache.org Subject: Kafka svn repo removed from incubator Hi, It seems that Kafka svn repository suddenly disappeared today. We officially graduated from incubator last week, but haven't filed infra tickets to move our repository yet. Is our repository automatically moved to somewhere? svn ls https://svn.apache.org/repos/asf/incubator/kafka svn: URL 'https://svn.apache.org/repos/asf/incubator/kafka' non-existent in that revision Thanks, Jun
[jira] [Created] (KAFKA-640) System Test Failures : kafka.common.InvalidClientIdException in broker log4j messages
John Fung created KAFKA-640:
---
Summary: System Test Failures :
kafka.common.InvalidClientIdException in broker log4j messages
Key: KAFKA-640
URL: https://issues.apache.org/jira/browse/KAFKA-640
Project: Kafka
Issue Type: Bug
Reporter: John Fung
* To reproduce the issue, download and build the latest Kafka 0.8 branch and
execute this command: kafka_home/system_test $ python -B
system_test_runner.py
* The following exception is found in the broker log4j messages in most System
Test cases:
[2012-11-29 09:06:21,322] WARN No previously checkpointed highwatermark value
found for topic test_1 partition 1. Returning 0 as the highwatermark
(kafka.server.HighwaterMarkCheckpoint)
[2012-11-29 09:06:21,326] INFO [Kafka Log on Broker 1], Truncated log segment
/tmp/kafka_server_1_logs/test_1-1/.log to target offset 0
(kafka.log.Log)
[2012-11-29 09:06:21,333] ERROR Replica Manager on Broker 1: Error processing
leaderAndISR request LeaderAndIsrRequest(1,,1000,Map((test_1,1) -
PartitionStateInfo(LeaderIsrAndControllerEpoch({
ISR:2,3,1,leader:2,leaderEpoch:0 },1),3), (test_1,0) -
PartitionStateInfo(LeaderIsrAndControllerEpoch({
ISR:1,2,3,leader:1,leaderEpoch:0
},1),3)),Set(id:2,creatorId:127.0.0.1-1354208764997,host:127.0.0.1,port:9092,
id:1,creatorId:127.0.0.1-1354208760105,host:127.0.0.1,port:9091),1)
(kafka.server.ReplicaManager)
kafka.common.InvalidClientIdException: ClientId
replica-fetcher-host_127.0.0.1-port_9092 is illegal, contains a character other
than ASCII alphanumerics, _ and -
at kafka.utils.ClientId$.validate(ClientIdAndTopic.scala:36)
at kafka.consumer.SimpleConsumer.init(SimpleConsumer.scala:81)
at
kafka.server.AbstractFetcherThread.init(AbstractFetcherThread.scala:44)
at
kafka.server.ReplicaFetcherThread.init(ReplicaFetcherThread.scala:26)
at
kafka.server.ReplicaFetcherManager.createFetcherThread(ReplicaFetcherManager.scala:26)
at
kafka.server.AbstractFetcherManager.addFetcher(AbstractFetcherManager.scala:44)
at kafka.cluster.Partition.makeFollower(Partition.scala:190)
at
kafka.server.ReplicaManager.kafka$server$ReplicaManager$$makeFollower(ReplicaManager.scala:236)
at
kafka.server.ReplicaManager$$anonfun$becomeLeaderOrFollower$3.apply(ReplicaManager.scala:201)
at
kafka.server.ReplicaManager$$anonfun$becomeLeaderOrFollower$3.apply(ReplicaManager.scala:191)
at scala.collection.immutable.Map$Map2.foreach(Map.scala:127)
at
kafka.server.ReplicaManager.becomeLeaderOrFollower(ReplicaManager.scala:191)
at kafka.server.KafkaApis.handleLeaderAndIsrRequest(KafkaApis.scala:129)
at kafka.server.KafkaApis.handle(KafkaApis.scala:60)
at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:41)
at java.lang.Thread.run(Thread.java:662)
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira
Re: Git
Yes, we are working on it. We will notify everyone once the git repo is ready. Thanks, Neha On Thu, Nov 29, 2012 at 7:25 AM, David Arthur mum...@gmail.com wrote: Has there ever been discussion of moving to Git? -David
[jira] [Commented] (KAFKA-133) Publish kafka jar to a public maven repository
[ https://issues.apache.org/jira/browse/KAFKA-133?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=13506824#comment-13506824 ] Otis Gospodnetic commented on KAFKA-133: There you have it: zkclient 0.2: http://search.maven.org/#artifactdetails%7Ccom.101tec%7Czkclient%7C0.2%7Cjar metrics 2.2.0: http://search.maven.org/#artifactdetails%7Ccom.yammer.metrics%7Cmetrics-parent%7C2.2.0%7Cpom This is now the most popular Kafka issue by far! Publish kafka jar to a public maven repository -- Key: KAFKA-133 URL: https://issues.apache.org/jira/browse/KAFKA-133 Project: Kafka Issue Type: Improvement Affects Versions: 0.6, 0.8 Reporter: Neha Narkhede Labels: patch Fix For: 0.8 Attachments: pom.xml The released kafka jar must be download manually and then deploy to a private repository before they can be used by a developer using maven2. Similar to other Apache projects, it will be nice to have a way to publish Kafka releases to a public maven repo. In the past, we gave it a try using sbt publish to Sonatype Nexus maven repo, but ran into some authentication problems. It will be good to revisit this and get it resolved. -- This message is automatically generated by JIRA. If you think it was sent incorrectly, please contact your JIRA administrators For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Commented] (KAFKA-633) AdminTest.testShutdownBroker fails
[ https://issues.apache.org/jira/browse/KAFKA-633?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=13506858#comment-13506858 ] Neha Narkhede commented on KAFKA-633: - That happens due to the test failure. If the test fails, it doesn't close down the zookeeper connections leading to these errors. Once we fix the root cause, these will disappear AdminTest.testShutdownBroker fails -- Key: KAFKA-633 URL: https://issues.apache.org/jira/browse/KAFKA-633 Project: Kafka Issue Type: Bug Components: core Affects Versions: 0.8 Reporter: Jun Rao 0m[ [31merror [0m] [0mTest Failed: testShutdownBroker(kafka.admin.AdminTest) [0m junit.framework.AssertionFailedError: expected:2 but was:3 at junit.framework.Assert.fail(Assert.java:47) at junit.framework.Assert.failNotEquals(Assert.java:277) at junit.framework.Assert.assertEquals(Assert.java:64) at junit.framework.Assert.assertEquals(Assert.java:195) at junit.framework.Assert.assertEquals(Assert.java:201) at kafka.admin.AdminTest.testShutdownBroker(AdminTest.scala:381) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25) at java.lang.reflect.Method.invoke(Method.java:597) at junit.framework.TestCase.runTest(TestCase.java:164) at junit.framework.TestCase.runBare(TestCase.java:130) at junit.framework.TestResult$1.protect(TestResult.java:110) at junit.framework.TestResult.runProtected(TestResult.java:128) at junit.framework.TestResult.run(TestResult.java:113) at junit.framework.TestCase.run(TestCase.java:120) at junit.framework.TestSuite.runTest(TestSuite.java:228) at junit.framework.TestSuite.run(TestSuite.java:223) at junit.framework.TestSuite.runTest(TestSuite.java:228) at junit.framework.TestSuite.run(TestSuite.java:223) at org.scalatest.junit.JUnit3Suite.run(JUnit3Suite.scala:309) at org.scalatest.tools.ScalaTestFramework$ScalaTestRunner.run(ScalaTestFramework.scala:40) at sbt.TestRunner.run(TestFramework.scala:53) at sbt.TestRunner.runTest$1(TestFramework.scala:67) at sbt.TestRunner.run(TestFramework.scala:76) at sbt.TestFramework$$anonfun$10$$anonfun$apply$11.runTest$2(TestFramework.scala:194) at sbt.TestFramework$$anonfun$10$$anonfun$apply$11$$anonfun$apply$12.apply(TestFramework.scala:205) at sbt.TestFramework$$anonfun$10$$anonfun$apply$11$$anonfun$apply$12.apply(TestFramework.scala:205) at sbt.NamedTestTask.run(TestFramework.scala:92) at sbt.ScalaProject$$anonfun$sbt$ScalaProject$$toTask$1.apply(ScalaProject.scala:193) at sbt.ScalaProject$$anonfun$sbt$ScalaProject$$toTask$1.apply(ScalaProject.scala:193) at sbt.TaskManager$Task.invoke(TaskManager.scala:62) at sbt.impl.RunTask.doRun$1(RunTask.scala:77) at sbt.impl.RunTask.runTask(RunTask.scala:85) at sbt.impl.RunTask.sbt$impl$RunTask$$runIfNotRoot(RunTask.scala:60) at sbt.impl.RunTask$$anonfun$runTasksExceptRoot$2.apply(RunTask.scala:48) at sbt.impl.RunTask$$anonfun$runTasksExceptRoot$2.apply(RunTask.scala:48) at sbt.Distributor$Run$Worker$$anonfun$2.apply(ParallelRunner.scala:131) at sbt.Distributor$Run$Worker$$anonfun$2.apply(ParallelRunner.scala:131) at sbt.Control$.trapUnit(Control.scala:19) at sbt.Distributor$Run$Worker.run(ParallelRunner.scala:131) -- This message is automatically generated by JIRA. If you think it was sent incorrectly, please contact your JIRA administrators For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Updated] (KAFKA-640) System Test Failures : kafka.common.InvalidClientIdException in broker log4j messages
[
https://issues.apache.org/jira/browse/KAFKA-640?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Swapnil Ghike updated KAFKA-640:
Attachment: kafka-640.patch
This happened because ClientId.validate(clientId) in SimpleConsumer did not
validate . in the clientId passed from ReplicaFetcherThread.
This patch fixes another bug - AbstractFetcherThread would create
SimpleConsumer and pass %s-host_%s-port_%s as the clientId to the
SimpleConsumer. SimpleConsumer would append the host-port string again to the
clientId while instantiating FetchRequestAndResponseStats.
Changes:
- The fix is to not include the host-port string while initializing
AbstractFetcherThread.clientId in ReplicaFetcherThread, the host and port are
already passed through the sourceBroker argument. This new clientId is passed
to SimpleConsumer in AbstractFetcherThread and it should validate successfully.
- Pass clientId + host-port string while instantiating *Stats in
AbstractFetcherThread and SimpleConsumer. Since the host-port string gives
information about the client, I think it should be ok to append it to clientId
and not create a separate case class like ClientIdAndTopic.
- Pass clientId + host-port string while instantiating FetchRequestBuilder in
AbstractFetcherThread.
- Pass clientId + host-port string to the constructors of ProducerRequestStats,
FetchRequestAndResponseStats, FetcherStats and FetcherLagStats to maintain
uniformity with passing clientId + host-port to FetchRequestBuilder in
AbstractFetcherThread.doWork().
- Removed a line in ClientIdTest.scala, it was redundant.
The validation criteria for clientId string that comes from the client is
unchanged. Ideally I would like to validate the clientId that *includes* the
host-port string, but that would require an introduction of '.' in the legal
characters set which would be inconsistent with legal chars for Topic. Instead,
we can maintain the same legal chars set and take care that the host-port
string doesn't change format within the code.
System Test Failures : kafka.common.InvalidClientIdException in broker log4j
messages
-
Key: KAFKA-640
URL: https://issues.apache.org/jira/browse/KAFKA-640
Project: Kafka
Issue Type: Bug
Reporter: John Fung
Labels: replication-testing
Attachments: kafka-640.patch
* To reproduce the issue, download and build the latest Kafka 0.8 branch and
execute this command: kafka_home/system_test $ python -B
system_test_runner.py
* The following exception is found in the broker log4j messages in most
System Test cases:
[2012-11-29 09:06:21,322] WARN No previously checkpointed highwatermark value
found for topic test_1 partition 1. Returning 0 as the highwatermark
(kafka.server.HighwaterMarkCheckpoint)
[2012-11-29 09:06:21,326] INFO [Kafka Log on Broker 1], Truncated log segment
/tmp/kafka_server_1_logs/test_1-1/.log to target offset 0
(kafka.log.Log)
[2012-11-29 09:06:21,333] ERROR Replica Manager on Broker 1: Error processing
leaderAndISR request LeaderAndIsrRequest(1,,1000,Map((test_1,1) -
PartitionStateInfo(LeaderIsrAndControllerEpoch({
ISR:2,3,1,leader:2,leaderEpoch:0 },1),3), (test_1,0) -
PartitionStateInfo(LeaderIsrAndControllerEpoch({
ISR:1,2,3,leader:1,leaderEpoch:0
},1),3)),Set(id:2,creatorId:127.0.0.1-1354208764997,host:127.0.0.1,port:9092,
id:1,creatorId:127.0.0.1-1354208760105,host:127.0.0.1,port:9091),1)
(kafka.server.ReplicaManager)
kafka.common.InvalidClientIdException: ClientId
replica-fetcher-host_127.0.0.1-port_9092 is illegal, contains a character
other than ASCII alphanumerics, _ and -
at kafka.utils.ClientId$.validate(ClientIdAndTopic.scala:36)
at kafka.consumer.SimpleConsumer.init(SimpleConsumer.scala:81)
at
kafka.server.AbstractFetcherThread.init(AbstractFetcherThread.scala:44)
at
kafka.server.ReplicaFetcherThread.init(ReplicaFetcherThread.scala:26)
at
kafka.server.ReplicaFetcherManager.createFetcherThread(ReplicaFetcherManager.scala:26)
at
kafka.server.AbstractFetcherManager.addFetcher(AbstractFetcherManager.scala:44)
at kafka.cluster.Partition.makeFollower(Partition.scala:190)
at
kafka.server.ReplicaManager.kafka$server$ReplicaManager$$makeFollower(ReplicaManager.scala:236)
at
kafka.server.ReplicaManager$$anonfun$becomeLeaderOrFollower$3.apply(ReplicaManager.scala:201)
at
kafka.server.ReplicaManager$$anonfun$becomeLeaderOrFollower$3.apply(ReplicaManager.scala:191)
at scala.collection.immutable.Map$Map2.foreach(Map.scala:127)
at
kafka.server.ReplicaManager.becomeLeaderOrFollower(ReplicaManager.scala:191)
at
[jira] [Assigned] (KAFKA-633) AdminTest.testShutdownBroker fails
[ https://issues.apache.org/jira/browse/KAFKA-633?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Joel Koshy reassigned KAFKA-633: Assignee: Joel Koshy AdminTest.testShutdownBroker fails -- Key: KAFKA-633 URL: https://issues.apache.org/jira/browse/KAFKA-633 Project: Kafka Issue Type: Bug Components: core Affects Versions: 0.8 Reporter: Jun Rao Assignee: Joel Koshy 0m[ [31merror [0m] [0mTest Failed: testShutdownBroker(kafka.admin.AdminTest) [0m junit.framework.AssertionFailedError: expected:2 but was:3 at junit.framework.Assert.fail(Assert.java:47) at junit.framework.Assert.failNotEquals(Assert.java:277) at junit.framework.Assert.assertEquals(Assert.java:64) at junit.framework.Assert.assertEquals(Assert.java:195) at junit.framework.Assert.assertEquals(Assert.java:201) at kafka.admin.AdminTest.testShutdownBroker(AdminTest.scala:381) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25) at java.lang.reflect.Method.invoke(Method.java:597) at junit.framework.TestCase.runTest(TestCase.java:164) at junit.framework.TestCase.runBare(TestCase.java:130) at junit.framework.TestResult$1.protect(TestResult.java:110) at junit.framework.TestResult.runProtected(TestResult.java:128) at junit.framework.TestResult.run(TestResult.java:113) at junit.framework.TestCase.run(TestCase.java:120) at junit.framework.TestSuite.runTest(TestSuite.java:228) at junit.framework.TestSuite.run(TestSuite.java:223) at junit.framework.TestSuite.runTest(TestSuite.java:228) at junit.framework.TestSuite.run(TestSuite.java:223) at org.scalatest.junit.JUnit3Suite.run(JUnit3Suite.scala:309) at org.scalatest.tools.ScalaTestFramework$ScalaTestRunner.run(ScalaTestFramework.scala:40) at sbt.TestRunner.run(TestFramework.scala:53) at sbt.TestRunner.runTest$1(TestFramework.scala:67) at sbt.TestRunner.run(TestFramework.scala:76) at sbt.TestFramework$$anonfun$10$$anonfun$apply$11.runTest$2(TestFramework.scala:194) at sbt.TestFramework$$anonfun$10$$anonfun$apply$11$$anonfun$apply$12.apply(TestFramework.scala:205) at sbt.TestFramework$$anonfun$10$$anonfun$apply$11$$anonfun$apply$12.apply(TestFramework.scala:205) at sbt.NamedTestTask.run(TestFramework.scala:92) at sbt.ScalaProject$$anonfun$sbt$ScalaProject$$toTask$1.apply(ScalaProject.scala:193) at sbt.ScalaProject$$anonfun$sbt$ScalaProject$$toTask$1.apply(ScalaProject.scala:193) at sbt.TaskManager$Task.invoke(TaskManager.scala:62) at sbt.impl.RunTask.doRun$1(RunTask.scala:77) at sbt.impl.RunTask.runTask(RunTask.scala:85) at sbt.impl.RunTask.sbt$impl$RunTask$$runIfNotRoot(RunTask.scala:60) at sbt.impl.RunTask$$anonfun$runTasksExceptRoot$2.apply(RunTask.scala:48) at sbt.impl.RunTask$$anonfun$runTasksExceptRoot$2.apply(RunTask.scala:48) at sbt.Distributor$Run$Worker$$anonfun$2.apply(ParallelRunner.scala:131) at sbt.Distributor$Run$Worker$$anonfun$2.apply(ParallelRunner.scala:131) at sbt.Control$.trapUnit(Control.scala:19) at sbt.Distributor$Run$Worker.run(ParallelRunner.scala:131) -- This message is automatically generated by JIRA. If you think it was sent incorrectly, please contact your JIRA administrators For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Commented] (KAFKA-633) AdminTest.testShutdownBroker fails
[ https://issues.apache.org/jira/browse/KAFKA-633?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=13506917#comment-13506917 ] Jay Kreps commented on KAFKA-633: - But that's a problem, no? Those test harnesses are supposed to clean up after themselves robustly--even if the test fails. AdminTest.testShutdownBroker fails -- Key: KAFKA-633 URL: https://issues.apache.org/jira/browse/KAFKA-633 Project: Kafka Issue Type: Bug Components: core Affects Versions: 0.8 Reporter: Jun Rao Assignee: Joel Koshy 0m[ [31merror [0m] [0mTest Failed: testShutdownBroker(kafka.admin.AdminTest) [0m junit.framework.AssertionFailedError: expected:2 but was:3 at junit.framework.Assert.fail(Assert.java:47) at junit.framework.Assert.failNotEquals(Assert.java:277) at junit.framework.Assert.assertEquals(Assert.java:64) at junit.framework.Assert.assertEquals(Assert.java:195) at junit.framework.Assert.assertEquals(Assert.java:201) at kafka.admin.AdminTest.testShutdownBroker(AdminTest.scala:381) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25) at java.lang.reflect.Method.invoke(Method.java:597) at junit.framework.TestCase.runTest(TestCase.java:164) at junit.framework.TestCase.runBare(TestCase.java:130) at junit.framework.TestResult$1.protect(TestResult.java:110) at junit.framework.TestResult.runProtected(TestResult.java:128) at junit.framework.TestResult.run(TestResult.java:113) at junit.framework.TestCase.run(TestCase.java:120) at junit.framework.TestSuite.runTest(TestSuite.java:228) at junit.framework.TestSuite.run(TestSuite.java:223) at junit.framework.TestSuite.runTest(TestSuite.java:228) at junit.framework.TestSuite.run(TestSuite.java:223) at org.scalatest.junit.JUnit3Suite.run(JUnit3Suite.scala:309) at org.scalatest.tools.ScalaTestFramework$ScalaTestRunner.run(ScalaTestFramework.scala:40) at sbt.TestRunner.run(TestFramework.scala:53) at sbt.TestRunner.runTest$1(TestFramework.scala:67) at sbt.TestRunner.run(TestFramework.scala:76) at sbt.TestFramework$$anonfun$10$$anonfun$apply$11.runTest$2(TestFramework.scala:194) at sbt.TestFramework$$anonfun$10$$anonfun$apply$11$$anonfun$apply$12.apply(TestFramework.scala:205) at sbt.TestFramework$$anonfun$10$$anonfun$apply$11$$anonfun$apply$12.apply(TestFramework.scala:205) at sbt.NamedTestTask.run(TestFramework.scala:92) at sbt.ScalaProject$$anonfun$sbt$ScalaProject$$toTask$1.apply(ScalaProject.scala:193) at sbt.ScalaProject$$anonfun$sbt$ScalaProject$$toTask$1.apply(ScalaProject.scala:193) at sbt.TaskManager$Task.invoke(TaskManager.scala:62) at sbt.impl.RunTask.doRun$1(RunTask.scala:77) at sbt.impl.RunTask.runTask(RunTask.scala:85) at sbt.impl.RunTask.sbt$impl$RunTask$$runIfNotRoot(RunTask.scala:60) at sbt.impl.RunTask$$anonfun$runTasksExceptRoot$2.apply(RunTask.scala:48) at sbt.impl.RunTask$$anonfun$runTasksExceptRoot$2.apply(RunTask.scala:48) at sbt.Distributor$Run$Worker$$anonfun$2.apply(ParallelRunner.scala:131) at sbt.Distributor$Run$Worker$$anonfun$2.apply(ParallelRunner.scala:131) at sbt.Control$.trapUnit(Control.scala:19) at sbt.Distributor$Run$Worker.run(ParallelRunner.scala:131) -- This message is automatically generated by JIRA. If you think it was sent incorrectly, please contact your JIRA administrators For more information on JIRA, see: http://www.atlassian.com/software/jira
Re: Protocol documentation draft
Okay here is a proposal to address the issues I raised. 1, 2. Correlation id. This is not strictly speaking needed, but it is maybe useful for debugging to be able to trace a particular request from client to server. So we will extend this across all the requests. 3. For metadata response I will try to fix this up by normalizing out the broker list and having the isr, replicas, and leader field just have the node id. 4. This should be uncontroversial and easy to add. 5. Let's remove creator id, it isn't used. 6. Let's generalize offset request. My proposal is below: Rename TopicMetadata API to ClusterMetadata, as this will contain all the data that is known cluster-wide. Then let's generalize the offset request to be PartitionMetadata--namely stuff about a particular partition on a particular server. The format of PartitionMetdata would be the following: PartitionMetadataRequest = [TopicName [PartitionId MinSegmentTime MaxSegmentInfos]] TopicName = string PartitionId = uint32 MinSegmentTime = uint64 MaxSegmentInfos = int32 PartitionMetadataResponse = [TopicName [PartitionMetadata]] TopicName = string PartitionMetadata = PartitionId LogSize NumberOfSegments LogEndOffset HighwaterMark [SegmentData] SegmentData = StartOffset LastModifiedTime LogSize = uint64 NumberOfSegments = int32 LogEndOffset = int64 HighwaterMark = int64 This would be general enough that we could continue to add to it for any new pieces of data we need. -Jay On Thu, Nov 29, 2012 at 3:17 PM, Jay Kreps jay.kr...@gmail.com wrote: I started trying to document the 0.8 protocol from the code and write a guide to client implementation. This is meant to be a more user-friendly and up-to-date version of the proposal wiki we had on the protocol changes. Here is what I wrote up so far: https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol I would love feedback on this document. It would be great if we could document everything you need to know to write a client so people don't need to reverse engineer our code. In doing this I found a number of things, some which I feel should be fixed in 0.8 some of which maybe can wait. 1. Correlation id is not used across all the requests. I don't think it can work as intended because of this. 2. On reflection I am not sure that we need a correlation id field. I think that since we need to guarantee that processing is sequential on any particular socket we can correlate with a simple queue. (e.g. as the client sends messages it adds them to a queue and as it receives responses it just correlates to whatever is at the head of the queue). 3. The metadata response seems to have a number of problems. Among them is that it weirdly repeats all the broker information many times. The response includes the ISR, leader (maybe), and the replicas. Each of these repeat all the broker information. This is super weird. I think what we should be doing here is including all broker information for all brokers and then just having the appropriate ids for the isr, leader, and replicas. 4. For topic discovery I think we need to support the case where no topics are specified in the metadata request and for this return information about all topics. I don't think we do this now. 5. I don't understand what the creator id is. 6. The offset request and response is not fully thought through and should be generalized. -Jay
