Ah good point. The replication factor at the beginning is 1! I'll rerun the
test with replication factor = 2.
On Mon, Jul 28, 2014 at 5:17 PM, Guozhang Wang <wangg...@gmail.com> wrote:
> Also with 2 brokers initially, is the replication factor also 2 at the
> beginning or just 1?
>
> On Mon, Jul 28, 2014 at 1:55 PM, Jad Naous <jad.na...@appdynamics.com>
> wrote:
>
> > The logs are quite large. I've sifted through them, and I'm attaching the
> > logs for the relevant parts where the lost message goes through the
> system.
> > Here's what the test does:
> >
> > 0) Start two brokers, one producer, one consumer. Topic has 20
> partitions,
> > using default partitioning scheme (which seems to send data to only a
> > couple of partitions when the keys are null, but that doesn't matter for
> > this test).
> > 1) Start a data generator sending data through Kafka continuously
> > 2) Start a new broker
> > 3) Reassign partitions: {"version": 1, "partitions":[
> > {"topic":"EventServiceUpsertTopic","partition":0, "replicas":
> > [0, 1, 2]},
> > {"topic":"EventServiceUpsertTopic","partition":1, "replicas":
> > [1, 2, 0]},
> > {"topic":"EventServiceUpsertTopic","partition":2, "replicas":
> > [2, 0, 1]},
> > {"topic":"EventServiceUpsertTopic","partition":3, "replicas":
> > [0, 1, 2]},
> > {"topic":"EventServiceUpsertTopic","partition":4, "replicas":
> > [1, 2, 0]},
> > {"topic":"EventServiceUpsertTopic","partition":5, "replicas":
> > [2, 0, 1]},
> > {"topic":"EventServiceUpsertTopic","partition":6, "replicas":
> > [0, 1, 2]},
> > {"topic":"EventServiceUpsertTopic","partition":7, "replicas":
> > [1, 2, 0]},
> > {"topic":"EventServiceUpsertTopic","partition":8, "replicas":
> > [2, 0, 1]},
> > {"topic":"EventServiceUpsertTopic","partition":9, "replicas":
> > [0, 1, 2]},
> > {"topic":"EventServiceUpsertTopic","partition":10, "replicas":
> > [1, 2, 0]},
> > {"topic":"EventServiceUpsertTopic","partition":11, "replicas":
> > [2, 0, 1]},
> > {"topic":"EventServiceUpsertTopic","partition":12, "replicas":
> > [0, 1, 2]},
> > {"topic":"EventServiceUpsertTopic","partition":13, "replicas":
> > [1, 2, 0]},
> > {"topic":"EventServiceUpsertTopic","partition":14, "replicas":
> > [2, 0, 1]},
> > {"topic":"EventServiceUpsertTopic","partition":15, "replicas":
> > [0, 1, 2]},
> > {"topic":"EventServiceUpsertTopic","partition":16, "replicas":
> > [1, 2, 0]},
> > {"topic":"EventServiceUpsertTopic","partition":17, "replicas":
> > [2, 0, 1]},
> > {"topic":"EventServiceUpsertTopic","partition":18, "replicas":
> > [0, 1, 2]},
> > {"topic":"EventServiceUpsertTopic","partition":19, "replicas":
> > [1, 2, 0]}]}
> > 4) Wait until reassignment is complete (i.e. until
> > ZkUtils.getPartitionsBeingReassigned() returns empty map)
> > 5) Wait until all replicas are caught up (i.e. until
> > ZkUtils.getLeaderAndIsrForPartition() returns all brokers in the ISR for
> > each partition)
> > 6) Trigger leader re-election by calling the
> > PreferredReplicaLeaderElectionCommand
> > 7) Wait until all the leaders are the preferred leaders for partitions
> > according to the replica reassignment from step 3
> > 8) Stop the data generator
> > 9) Check that all the data was consumed
> >
> > You can seem from the producer.log that the data: {"field1": ["10"],
> > "idField": "id-5-59"} was sent to broker0 successfully, but the consumer
> > never sees it.
> >
> > Thanks,
> > Jad.
> >
> >
> >
> > On Mon, Jul 28, 2014 at 10:33 AM, Guozhang Wang <wangg...@gmail.com>
> > wrote:
> >
> >> Also, for the initial two replica case did you see any error/warning
> logs
> >> on your producers?
> >>
> >> Guozhang
> >>
> >>
> >> On Mon, Jul 28, 2014 at 10:32 AM, Guozhang Wang <wangg...@gmail.com>
> >> wrote:
> >>
> >> > Hi Jad,
> >> >
> >> > Just to clarify, you also see data loss when you created the topic
> with
> >> > replica factor 2, and two replicas running, and after an auto leader
> >> > election triggered? If that is the case could you attach the logs of
> all
> >> > involved brokers here?
> >> >
> >> > For your second question, KAFKA-1211 is designed to handle that case.
> >> >
> >> > Guozhang
> >> >
> >> >
> >> > On Mon, Jul 28, 2014 at 10:18 AM, Jad Naous <
> jad.na...@appdynamics.com>
> >> > wrote:
> >> >
> >> >> Guozhang,
> >> >>
> >> >> I have actually also seen this happen when there are two replicas
> >> >> initially. So this problem is not limited to 1 replica. The issue is
> >> the
> >> >> truncation after leader election, which will also happen on the
> second
> >> >> replica.
> >> >>
> >> >> Coming back to your objections:
> >> >>
> >> >>
> >> >> First case: inconsistency between replicas
> >> >> > 1) currently replica 1 is the current leader
> >> >> > replica 1: m1 m2 m3
> >> >> > replica 2: m1 m2
> >> >> > 2) replica 1 fails, and replica 2 becomes the new leader and accept
> >> >> > messages 4 and 5:
> >> >> > replica 1: m1 m2 m3
> >> >> > replica 2: m1 m2 m4 m5
> >> >> > 3) replica 1 resumes, and does not truncate to HW, then it will
> still
> >> >> > maintain m3, which is actually never "committed". Say leader moves
> to
> >> >> > replica 1 again, we can ended up with:
> >> >> > replica 1: m1 m2 m3 m6
> >> >> > replica 2: m1 m2 m4 m5
> >> >>
> >> >>
> >> >> I see. So when a replica resumes, it has to truncate to the last HW
> it
> >> saw
> >> >> before it died.
> >> >>
> >> >>
> >> >> >
> >> >> > Second case: inconsistency between server and clients:
> >> >> > 1) producer send message m3 with ack=-1:
> >> >> > replica 1: m1 m2 m3
> >> >> > replica 2: m1 m2 m3
> >> >> > replica 3: m1 m2
> >> >> > 2) the response is held until all replicas also gets m3, say at
> this
> >> >> time
> >> >> > current leader replica 1 fails and replica 3 re-elects. If replica
> 3
> >> >> gets
> >> >> > up to the largest LEO it will also get m3.
> >> >> > replica 2: m1 m2 m3
> >> >> > replica 3: m1 m2 m3
> >> >> > 3) But m3 is not actually "committed" by the time replica 1 fails;
> >> when
> >> >> > producer gets the error at the time replica 1 fails, it will think
> >> that
> >> >> m3
> >> >> > was not successfully sent, so retry sending m3:
> >> >> > replica 2: m1 m2 m3 m3
> >> >> > replica 3: m1 m2 m3 m3
> >> >>
> >> >>
> >> >> So on failure, all nodes need to truncate to the HW. But if there's
> no
> >> >> failure, then truncating would lose data unnecessarily. Maybe those
> two
> >> >> scenarios need to be handled differently?
> >> >>
> >> >> Jad.
> >> >>
> >> >>
> >> >> On Mon, Jul 28, 2014 at 9:58 AM, Guozhang Wang <wangg...@gmail.com>
> >> >> wrote:
> >> >>
> >> >> > Jun, Jad,
> >> >> >
> >> >> > I think in this case data loss can still happen, since the replica
> >> >> factor
> >> >> > was previously one, and in handling the produce requests, if the
> >> server
> >> >> > decides that all the produced partitions have a replica factor of 1
> >> it
> >> >> will
> >> >> > also directly send back the response instead of putting the request
> >> into
> >> >> > purgatory even if currently the number of replicas is 2 (for
> details
> >> >> look
> >> >> > at ReplicaManager.getReplicationFactorForPartition and search of
> the
> >> >> usage
> >> >> > of Partition.replicationFactor).
> >> >> >
> >> >> > I now agree that this is not related to KAFKA-1211 but a different
> >> small
> >> >> > bug. We need to probably file another JIRA for this. But I think
> >> after
> >> >> this
> >> >> > one is fixed (which should be much easier than KAFKA-1211), Jad's
> >> >> scenario
> >> >> > should not cause data loss anymore.
> >> >> >
> >> >> > Guozhang
> >> >> >
> >> >> >
> >> >> > On Sun, Jul 27, 2014 at 6:11 PM, Jad Naous <
> >> jad.na...@appdynamics.com>
> >> >> > wrote:
> >> >> >
> >> >> > > So in summary, is it true to say that currently triggering leader
> >> >> > > reelection is not a safe operation? I have been able to reproduce
> >> that
> >> >> > > message loss pretty reliably in tests. If that is the case, isn't
> >> >> that an
> >> >> > > important operation in a large cluster where nodes go up and
> down?
> >> >> > > On Jul 25, 2014 10:00 PM, "Jun Rao" <jun...@gmail.com> wrote:
> >> >> > >
> >> >> > > > Actually, I don't think KAFKA-1211 will happen with just 2
> >> replicas.
> >> >> > > When a
> >> >> > > > replica becomes a leader, it never truncates its log. Only
> when a
> >> >> > replica
> >> >> > > > becomes follower, it truncates its log to HW. So in this
> >> particular
> >> >> > case,
> >> >> > > > the new leader will not truncate data to offset 8.
> >> >> > > >
> >> >> > > > Thanks,
> >> >> > > >
> >> >> > > > Jun
> >> >> > > >
> >> >> > > >
> >> >> > > > On Fri, Jul 25, 2014 at 3:37 PM, Guozhang Wang <
> >> wangg...@gmail.com>
> >> >> > > wrote:
> >> >> > > >
> >> >> > > > > Hi Jad,
> >> >> > > > >
> >> >> > > > > Yes. In this case I think you are actually hitting
> KAFKA-1211.
> >> The
> >> >> > > > summary
> >> >> > > > > of the issue is that, it takes one more fetch request round
> >> trip
> >> >> for
> >> >> > > the
> >> >> > > > > follower replica to advance the HW after the leader has
> >> advanced
> >> >> HW.
> >> >> > So
> >> >> > > > for
> >> >> > > > > your case, the whole process is like this:
> >> >> > > > >
> >> >> > > > > 1. leader LEO at 10, follower LEO at 8. Both leader and
> >> follower
> >> >> > knows
> >> >> > > > the
> >> >> > > > > LEO is at 8.
> >> >> > > > > 2. Follower fetch data on Leader starting at 8, leader
> records
> >> its
> >> >> > LEO
> >> >> > > as
> >> >> > > > > 8.
> >> >> > > > > 3. Follower gets 9 and 10 and append to its local log.
> >> >> > > > > 4. Follower fetch data on Leader starting at 10, leader
> records
> >> >> its
> >> >> > LEO
> >> >> > > > as
> >> >> > > > > 10; now leader knows follower has caught up, it advances its
> HW
> >> >> to 10
> >> >> > > and
> >> >> > > > > adds the follower to ISR (but follower does not know that
> yet!
> >> It
> >> >> > still
> >> >> > > > > think the HW is 8).
> >> >> > > > > 5. Leader's fetch response gets back to follower, and now the
> >> >> > follower
> >> >> > > > > knows that HW has been updated to 10.
> >> >> > > > >
> >> >> > > > > And let's say there is a leader election between step 4) and
> >> 5),
> >> >> for
> >> >> > > your
> >> >> > > > > case it is due to preferred leader election, but it could
> also
> >> be
> >> >> > that
> >> >> > > > > current leader fails, etc. Then on becoming the new leader
> the
> >> >> > follower
> >> >> > > > > will truncate its data to 8, which is the HW it knows. Hence
> >> the
> >> >> data
> >> >> > > > loss.
> >> >> > > > >
> >> >> > > > > The proposed solution in KAFKA-1211 will tackle this issue.
> >> >> > > > >
> >> >> > > > > Guozhang
> >> >> > > > >
> >> >> > > > >
> >> >> > > > > On Fri, Jul 25, 2014 at 2:48 PM, Jad Naous <
> >> >> > jad.na...@appdynamics.com>
> >> >> > > > > wrote:
> >> >> > > > >
> >> >> > > > > > Hi Guozhang,
> >> >> > > > > >
> >> >> > > > > > Yes, broker 1 is in the ISR (in fact, I wait for broker 1
> to
> >> be
> >> >> in
> >> >> > > the
> >> >> > > > > ISR
> >> >> > > > > > before triggering election). However, I think there is
> >> something
> >> >> > > still
> >> >> > > > > > amiss. I still see data loss. Here are some relevant log
> >> lines
> >> >> from
> >> >> > > > > broker
> >> >> > > > > > 0 (different test run). The log on broker 0 is getting
> >> >> truncated,
> >> >> > > > losing
> >> >> > > > > > some messages.
> >> >> > > > > >
> >> >> > > > > > [2014-07-25 10:40:02,134] [DEBUG]
> [kafka-request-handler-5]
> >> >> > > > > > [kafka.cluster.Partition] Partition
> >> >> [EventServiceUpsertTopic,19]
> >> >> > on
> >> >> > > > > broker
> >> >> > > > > > 0: Old hw for partition [EventServiceUpsertTopic,19] is
> 8082.
> >> >> New
> >> >> > hw
> >> >> > > is
> >> >> > > > > > 8082. All leo's are 8111,8082
> >> >> > > > > >
> >> >> > > > > > [2014-07-25 10:40:02,134] [DEBUG]
> [kafka-request-handler-5]
> >> >> > > > > > [kafka.server.KafkaApis] [KafkaApi-0] Produce to local log
> >> in
> >> >> 1 ms
> >> >> > > > > >
> >> >> > > > > > [2014-07-25 10:40:02,134] [DEBUG]
> >> >> > > [main-SendThread(localhost:49893)]
> >> >> > > > > > [org.apache.zookeeper.ClientCnxn] Reading reply
> >> >> > > > > > sessionid:0x476e9a9e9a0001, packet:: clientPath:null
> >> >> > serverPath:null
> >> >> > > > > > finished:false header:: 729,5 replyHeader::
> 729,4294968217,0
> >> >> > > > request::
> >> >> > > > > >
> >> >> > > > > >
> >> >> > > > >
> >> >> > > >
> >> >> > >
> >> >> >
> >> >>
> >>
> '/brokers/topics/EventServiceUpsertTopic/partitions/5/state,#7b22636f6e74726f6c6c65725f65706f6368223a312c226c6561646572223a312c2276657273696f6e223a312c226c65616465725f65706f6368223a332c22697372223a5b302c315d7d,3
> >> >> > > > > > response::
> >> >> > > > > >
> >> >> > > > > >
> >> >> > > > >
> >> >> > > >
> >> >> > >
> >> >> >
> >> >>
> >>
> s{4294967416,4294968217,1406309966419,1406310002132,4,0,0,0,74,0,4294967416}
> >> >> > > > > >
> >> >> > > > > > [2014-07-25 10:40:02,134] [DEBUG]
> [kafka-processor-49917-2]
> >> >> > > > > > [kafka.request.logger] Completed request:Name:
> >> ProducerRequest;
> >> >> > > > Version:
> >> >> > > > > > 0; CorrelationId: 16248; ClientId: ; RequiredAcks: -1;
> >> >> > AckTimeoutMs:
> >> >> > > > > 10000
> >> >> > > > > > ms from client /127.0.0.1:50168
> >> >> > > > > >
> >> >> > > > > >
> >> >> > > > >
> >> >> > > >
> >> >> > >
> >> >> >
> >> >>
> >>
> ;totalTime:1,requestQueueTime:0,localTime:1,remoteTime:0,responseQueueTime:0,sendTime:0
> >> >> > > > > >
> >> >> > > > > > [2014-07-25 10:40:02,134] [DEBUG]
> >> >> > > > > >
> >> >> > > >
> >> >> >
> >> >>
> >>
> [ZkClient-EventThread-13-localhost:49893,localhost:49896,localhost:49899]
> >> >> > > > > > [kafka.utils.ZkUtils$] Conditional update of path
> >> >> > > > > > /brokers/topics/EventServiceUpsertTopic/partitions/5/state
> >> with
> >> >> > value
> >> >> > > > > >
> >> >> > > > >
> >> >> > > >
> >> >> > >
> >> >> >
> >> >>
> >>
> {"controller_epoch":1,"leader":1,"version":1,"leader_epoch":3,"isr":[0,1]}
> >> >> > > > > > and expected version 3 succeeded, returning the new
> version:
> >> 4
> >> >> > > > > >
> >> >> > > > > > [2014-07-25 10:40:02,136] [DEBUG]
> >> >> > > > > >
> >> >> > > >
> >> >> >
> >> >>
> >>
> [ZkClient-EventThread-13-localhost:49893,localhost:49896,localhost:49899]
> >> >> > > > > > [k.controller.PartitionStateMachine] [Partition state
> >> machine
> >> >> on
> >> >> > > > > > Controller 0]: After leader election, leader cache is
> >> updated to
> >> >> > > > > > Map([EventServiceUpsertTopic,18] ->
> >> >> > > > > > (Leader:0,ISR:0,LeaderEpoch:0,ControllerEpoch:1),
> >> >> > > > > > [EventServiceUpsertTopic,4] ->
> >> >> > > > > > (Leader:0,ISR:0,LeaderEpoch:0,ControllerEpoch:1),
> >> >> > > > > > [EventServiceUpsertTopic,7] ->
> >> >> > > > > > (Leader:1,ISR:0,1,LeaderEpoch:3,ControllerEpoch:1),
> >> >> > > > > > [EventServiceUpsertTopic,17] ->
> >> >> > > > > > (Leader:1,ISR:0,1,LeaderEpoch:3,ControllerEpoch:1),
> >> >> > > > > > [EventServiceUpsertTopic,16] ->
> >> >> > > > > > (Leader:0,ISR:0,LeaderEpoch:0,ControllerEpoch:1),
> >> >> > > > > > [EventServiceUpsertTopic,5] ->
> >> >> > > > > > (Leader:1,ISR:0,1,LeaderEpoch:3,ControllerEpoch:1),
> >> >> > > > > > [EventServiceUpsertTopic,10] ->
> >> >> > > > > > (Leader:0,ISR:0,LeaderEpoch:0,ControllerEpoch:1),
> >> >> > > > > > [EventServiceUpsertTopic,1] ->
> >> >> > > > > > (Leader:1,ISR:0,1,LeaderEpoch:3,ControllerEpoch:1),
> >> >> > > > > > [EventServiceUpsertTopic,13] ->
> >> >> > > > > > (Leader:1,ISR:0,1,LeaderEpoch:3,ControllerEpoch:1),
> >> >> > > > > > [EventServiceUpsertTopic,9] ->
> >> >> > > > > > (Leader:1,ISR:0,1,LeaderEpoch:3,ControllerEpoch:1),
> >> >> > > > > > [EventServiceUpsertTopic,8] ->
> >> >> > > > > > (Leader:0,ISR:0,LeaderEpoch:0,ControllerEpoch:1),
> >> >> > > > > > [EventServiceUpsertTopic,11] ->
> >> >> > > > > > (Leader:1,ISR:0,1,LeaderEpoch:3,ControllerEpoch:1),
> >> >> > > > > > [EventServiceUpsertTopic,6] ->
> >> >> > > > > > (Leader:0,ISR:0,LeaderEpoch:0,ControllerEpoch:1),
> >> >> > > > > > [EventServiceUpsertTopic,12] ->
> >> >> > > > > > (Leader:0,ISR:0,LeaderEpoch:0,ControllerEpoch:1),
> >> >> > > > > > [EventServiceUpsertTopic,2] ->
> >> >> > > > > > (Leader:0,ISR:0,LeaderEpoch:0,ControllerEpoch:1),
> >> >> > > > > > [EventServiceUpsertTopic,3] ->
> >> >> > > > > > (Leader:1,ISR:0,1,LeaderEpoch:3,ControllerEpoch:1),
> >> >> > > > > > [EventServiceUpsertTopic,14] ->
> >> >> > > > > > (Leader:0,ISR:0,LeaderEpoch:0,ControllerEpoch:1), [
> >> >> > > > > > EventServiceUpsertTopic,19] ->
> >> >> > > > > > (Leader:1,ISR:0,1,LeaderEpoch:3,ControllerEpoch:1),
> >> >> > > > > > [EventServiceUpsertTopic,0] ->
> >> >> > > > > > (Leader:0,ISR:0,LeaderEpoch:0,ControllerEpoch:1),
> >> >> > > > > > [EventServiceUpsertTopic,15] ->
> >> >> > > > > > (Leader:1,ISR:0,1,LeaderEpoch:3,ControllerEpoch:1))
> >> >> > > > > >
> >> >> > > > > > [2014-07-25 10:40:02,136] [DEBUG]
> [kafka-request-handler-1]
> >> >> > > > > > [kafka.cluster.Partition] Partition
> >> >> [EventServiceUpsertTopic,19]
> >> >> > on
> >> >> > > > > broker
> >> >> > > > > > 0: Old hw for partition [EventServiceUpsertTopic,19] is
> 8082.
> >> >> New
> >> >> > hw
> >> >> > > is
> >> >> > > > > > 8082. All leo's are 8112,8082
> >> >> > > > > >
> >> >> > > > > > [2014-07-25 10:40:02,136] [DEBUG]
> [kafka-request-handler-1]
> >> >> > > > > > [kafka.server.KafkaApis] [KafkaApi-0] Produce to local log
> >> in
> >> >> 0 ms
> >> >> > > > > >
> >> >> > > > > > [2014-07-25 10:40:02,137] [DEBUG]
> [kafka-processor-49917-2]
> >> >> > > > > > [kafka.request.logger] Completed request:Name:
> >> ProducerRequest;
> >> >> > > > Version:
> >> >> > > > > > 0; CorrelationId: 16250; ClientId: ; RequiredAcks: -1;
> >> >> > AckTimeoutMs:
> >> >> > > > > 10000
> >> >> > > > > > ms from client /127.0.0.1:50168
> >> >> > > > > >
> >> >> > > > > >
> >> >> > > > >
> >> >> > > >
> >> >> > >
> >> >> >
> >> >>
> >>
> ;totalTime:1,requestQueueTime:0,localTime:1,remoteTime:0,responseQueueTime:0,sendTime:0
> >> >> > > > > >
> >> >> > > > > > [2014-07-25 10:40:02,140] [INFO ]
> >> >> > > > > >
> >> >> > > >
> >> >> >
> >> >>
> >>
> [ZkClient-EventThread-13-localhost:49893,localhost:49896,localhost:49899]
> >> >> > > > > > [kafka.controller.KafkaController] [Controller 0]:
> >> Partition [
> >> >> > > > > > EventServiceUpsertTopic,19] completed preferred replica
> >> leader
> >> >> > > > election.
> >> >> > > > > > New leader is 1
> >> >> > > > > >
> >> >> > > > > > [2014-07-25 10:40:02,140] [INFO ]
> >> >> > > > > >
> >> >> > > >
> >> >> >
> >> >>
> >>
> [ZkClient-EventThread-13-localhost:49893,localhost:49896,localhost:49899]
> >> >> > > > > > [kafka.controller.KafkaController] [Controller 0]:
> Partition
> >> >> > > > > > [EventServiceUpsertTopic,5] completed preferred replica
> >> leader
> >> >> > > > election.
> >> >> > > > > > New leader is 1
> >> >> > > > > >
> >> >> > > > > > [2014-07-25 10:40:02,142] [DEBUG]
> >> >> > > [main-SendThread(localhost:49893)]
> >> >> > > > > > [org.apache.zookeeper.ClientCnxn] Got notification
> >> >> > > > > > sessionid:0x476e9a9e9a0001
> >> >> > > > > >
> >> >> > > > > > [2014-07-25 10:40:02,142] [DEBUG]
> >> >> > > [main-SendThread(localhost:49893)]
> >> >> > > > > > [org.apache.zookeeper.ClientCnxn] Got WatchedEvent
> >> >> > > state:SyncConnected
> >> >> > > > > > type:NodeDeleted path:/admin/preferred_replica_election for
> >> >> > sessionid
> >> >> > > > > > 0x476e9a9e9a0001
> >> >> > > > > >
> >> >> > > > > > [2014-07-25 10:40:02,143] [DEBUG]
> >> >> > > [main-SendThread(localhost:49893)]
> >> >> > > > > > [org.apache.zookeeper.ClientCnxn] Reading reply
> >> >> > > > > > sessionid:0x476e9a9e9a0001, packet:: clientPath:null
> >> >> > serverPath:null
> >> >> > > > > > finished:false header:: 730,2 replyHeader::
> 730,4294968218,0
> >> >> > > > request::
> >> >> > > > > > '/admin/preferred_replica_election,-1 response:: null
> >> >> > > > > >
> >> >> > > > > > [2014-07-25 10:40:02,145] [INFO ]
> [kafka-request-handler-0]
> >> >> > > > > > [kafka.server.ReplicaFetcherManager]
> [ReplicaFetcherManager
> >> on
> >> >> > > broker
> >> >> > > > 0]
> >> >> > > > > > Removed fetcher for partitions
> >> >> > > > > >
> >> >> > > > > >
> >> >> > > > >
> >> >> > > >
> >> >> > >
> >> >> >
> >> >>
> >>
> [EventServiceUpsertTopic,13],[EventServiceUpsertTopic,11],[EventServiceUpsertTopic,17],[EventServiceUpsertTopic,7],[EventServiceUpsertTopic,9],[EventServiceUpsertTopic,1],[EventServiceUpsertTopic,15],[
> >> >> > > > > > EventServiceUpsertTopic,19
> >> >> > > > > > ],[EventServiceUpsertTopic,3],[EventServiceUpsertTopic,5]
> >> >> > > > > >
> >> >> > > > > > [2014-07-25 10:40:02,153] [INFO ]
> [kafka-request-handler-0]
> >> >> > > > > > [kafka.log.Log] Truncating log EventServiceUpsertTopic-19
> to
> >> >> > offset
> >> >> > > > > 8082.
> >> >> > > > > >
> >> >> > > > > > [2014-07-25 10:40:02,159] [INFO ]
> [kafka-request-handler-0]
> >> >> > > > > > [kafka.log.Log] Truncating log EventServiceUpsertTopic-5
> to
> >> >> offset
> >> >> > > 0.
> >> >> > > > > >
> >> >> > > > > >
> >> >> > > > > > I have removed some irrelevant lines. As you can see, the
> log
> >> >> for
> >> >> > > > > > EventServiceUpsertTopic-19 is being truncated, losing
> >> messages
> >> >> that
> >> >> > > > have
> >> >> > > > > > not yet been replicated to broker 1. I'm not sure exactly
> >> what
> >> >> the
> >> >> > > > issue
> >> >> > > > > > is. Maybe new requests should be completely blocked after
> >> leader
> >> >> > > > election
> >> >> > > > > > until the new leader catches up to the messages that have
> >> been
> >> >> > acked
> >> >> > > > and
> >> >> > > > > it
> >> >> > > > > > has not yet received.
> >> >> > > > > >
> >> >> > > > > > Thanks,
> >> >> > > > > >
> >> >> > > > > > Jad.
> >> >> > > > > >
> >> >> > > > > >
> >> >> > > > > > On Fri, Jul 25, 2014 at 2:14 PM, Guozhang Wang <
> >> >> wangg...@gmail.com
> >> >> > >
> >> >> > > > > wrote:
> >> >> > > > > >
> >> >> > > > > > > Hello Jad,
> >> >> > > > > > >
> >> >> > > > > > > I double-checked the source code again, and found that
> >> >> actually
> >> >> > the
> >> >> > > > > > > preferred leader elector does consider whether the
> selected
> >> >> > replica
> >> >> > > > is
> >> >> > > > > in
> >> >> > > > > > > ISR or not. This means that by the time the election is
> >> >> > triggered,
> >> >> > > > > > broker 1
> >> >> > > > > > > is added to ISR by broker 0. Could you check before step
> >> 3, is
> >> >> > > there
> >> >> > > > > any
> >> >> > > > > > > log entries on broker 0 adding broker 1 to ISR or
> updating
> >> HW
> >> >> to
> >> >> > > 975?
> >> >> > > > > > >
> >> >> > > > > > >
> >> >> > > > > > > On Thu, Jul 24, 2014 at 4:21 PM, Jad Naous <
> >> >> > > > jad.na...@appdynamics.com>
> >> >> > > > > > > wrote:
> >> >> > > > > > >
> >> >> > > > > > > > Hi,
> >> >> > > > > > > >
> >> >> > > > > > > > I have a test that continuously sends messages to one
> >> >> broker,
> >> >> > > > brings
> >> >> > > > > up
> >> >> > > > > > > > another broker, and adds it as a replica for all
> >> partitions,
> >> >> > with
> >> >> > > > it
> >> >> > > > > > > being
> >> >> > > > > > > > the preferred replica for some. I have
> >> >> > > > > > auto.leader.rebalance.enable=true,
> >> >> > > > > > > > so replica election gets triggered. Data is being
> pumped
> >> to
> >> >> the
> >> >> > > old
> >> >> > > > > > > broker
> >> >> > > > > > > > all the while. It seems that some data gets lost while
> >> >> > switching
> >> >> > > > over
> >> >> > > > > > to
> >> >> > > > > > > > the new leader. Is this a bug, or do I have something
> >> >> > > > misconfigured?
> >> >> > > > > I
> >> >> > > > > > > also
> >> >> > > > > > > > have request.required.acks=-1 on the producer.
> >> >> > > > > > > >
> >> >> > > > > > > > Here's what I think is happening:
> >> >> > > > > > > >
> >> >> > > > > > > > 1. Producer writes message to broker 0,
> >> >> > > > [EventServiceUpsertTopic,13],
> >> >> > > > > > w/
> >> >> > > > > > > > broker 0 currently leader, with ISR=(0), so write
> returns
> >> >> > > > > successfully,
> >> >> > > > > > > > even when acks = -1. Correlation id 35836
> >> >> > > > > > > >
> >> >> > > > > > > > Producer log:
> >> >> > > > > > > >
> >> >> > > > > > > > [2014-07-24 14:44:26,991] [DEBUG] [dw-97 - PATCH
> >> >> > > > > > > >
> >> >> > > > > > > >
> >> >> > > > > > >
> >> >> > > > > >
> >> >> > > > >
> >> >> > > >
> >> >> > >
> >> >> >
> >> >>
> >>
> /v1/events/type_for_test_bringupNewBroker_shouldRebalance_shouldNotLoseData/event?_idPath=idField&_mergeFields=field1]
> >> >> > > > > > > > [kafka.producer.BrokerPartitionInfo] Partition
> >> >> > > > > > > > [EventServiceUpsertTopic,13] has leader 0
> >> >> > > > > > > >
> >> >> > > > > > > > [2014-07-24 14:44:26,993] [DEBUG] [dw-97 - PATCH
> >> >> > > > > > > >
> >> >> > > > > > > >
> >> >> > > > > > >
> >> >> > > > > >
> >> >> > > > >
> >> >> > > >
> >> >> > >
> >> >> >
> >> >>
> >>
> /v1/events/type_for_test_bringupNewBroker_shouldRebalance_shouldNotLoseData/event?_idPath=idField&_mergeFields=field1]
> >> >> > > > > > > > [k.producer.async.DefaultEventHandler] Producer sent
> >> >> messages
> >> >> > > with
> >> >> > > > > > > > correlation id 35836 for topics
> >> >> [EventServiceUpsertTopic,13] to
> >> >> > > > > broker
> >> >> > > > > > 0
> >> >> > > > > > > on
> >> >> > > > > > > > localhost:56821
> >> >> > > > > > > > 2. Broker 1 is still catching up
> >> >> > > > > > > >
> >> >> > > > > > > > Broker 0 Log:
> >> >> > > > > > > >
> >> >> > > > > > > > [2014-07-24 14:44:26,992] [DEBUG]
> >> >> [kafka-request-handler-3]
> >> >> > > > > > > > [kafka.cluster.Partition] Partition
> >> >> > [EventServiceUpsertTopic,13]
> >> >> > > > on
> >> >> > > > > > > broker
> >> >> > > > > > > > 0: Old hw for partition [EventServiceUpsertTopic,13] is
> >> 971.
> >> >> > New
> >> >> > > hw
> >> >> > > > > is
> >> >> > > > > > > 971.
> >> >> > > > > > > > All leo's are 975,971
> >> >> > > > > > > >
> >> >> > > > > > > > [2014-07-24 14:44:26,992] [DEBUG]
> >> >> [kafka-request-handler-3]
> >> >> > > > > > > > [kafka.server.KafkaApis] [KafkaApi-0] Produce to local
> >> log
> >> >> in
> >> >> > 0
> >> >> > > ms
> >> >> > > > > > > >
> >> >> > > > > > > > [2014-07-24 14:44:26,992] [DEBUG]
> >> >> [kafka-processor-56821-0]
> >> >> > > > > > > > [kafka.request.logger] Completed request:Name:
> >> >> > ProducerRequest;
> >> >> > > > > > Version:
> >> >> > > > > > > > 0; CorrelationId: 35836; ClientId: ; RequiredAcks: -1;
> >> >> > > > AckTimeoutMs:
> >> >> > > > > > > 10000
> >> >> > > > > > > > ms from client /127.0.0.1:57086
> >> >> > > > > > > >
> >> >> > > > > > > >
> >> >> > > > > > >
> >> >> > > > > >
> >> >> > > > >
> >> >> > > >
> >> >> > >
> >> >> >
> >> >>
> >>
> ;totalTime:0,requestQueueTime:0,localTime:0,remoteTime:0,responseQueueTime:0,sendTime:0
> >> >> > > > > > > > 3. Leader election is triggered by the scheduler:
> >> >> > > > > > > >
> >> >> > > > > > > > Broker 0 Log:
> >> >> > > > > > > >
> >> >> > > > > > > > [2014-07-24 14:44:26,991] [INFO ] [kafka-scheduler-0]
> >> >> > > > > > > > [k.c.PreferredReplicaPartitionLeaderSelector]
> >> >> > > > > > > > [PreferredReplicaPartitionLeaderSelector]: Current
> >> leader 0
> >> >> for
> >> >> > > > > > > partition [
> >> >> > > > > > > > EventServiceUpsertTopic,13] is not the preferred
> replica.
> >> >> > > > Trigerring
> >> >> > > > > > > > preferred replica leader election
> >> >> > > > > > > >
> >> >> > > > > > > > [2014-07-24 14:44:26,993] [DEBUG] [kafka-scheduler-0]
> >> >> > > > > > > > [kafka.utils.ZkUtils$] Conditional update of path
> >> >> > > > > > > >
> >> /brokers/topics/EventServiceUpsertTopic/partitions/13/state
> >> >> > with
> >> >> > > > > value
> >> >> > > > > > > >
> >> >> > > > > > >
> >> >> > > > > >
> >> >> > > > >
> >> >> > > >
> >> >> > >
> >> >> >
> >> >>
> >>
> {"controller_epoch":1,"leader":1,"version":1,"leader_epoch":3,"isr":[0,1]}
> >> >> > > > > > > > and expected version 3 succeeded, returning the new
> >> >> version: 4
> >> >> > > > > > > >
> >> >> > > > > > > > [2014-07-24 14:44:26,994] [DEBUG] [kafka-scheduler-0]
> >> >> > > > > > > > [k.controller.PartitionStateMachine] [Partition state
> >> >> machine
> >> >> > on
> >> >> > > > > > > > Controller 0]: After leader election, leader cache is
> >> >> updated
> >> >> > to
> >> >> > > > > > > >
> >> >> > > > > > >
> >> >> > > > > >
> >> >> > > > >
> >> >> > > >
> >> >> > >
> >> >> >
> >> >>
> >>
> Map(<Snipped>(Leader:1,ISR:0,1,LeaderEpoch:3,ControllerEpoch:1),<EndSnip>)
> >> >> > > > > > > >
> >> >> > > > > > > > [2014-07-24 14:44:26,994] [INFO ] [kafka-scheduler-0]
> >> >> > > > > > > > [kafka.controller.KafkaController] [Controller 0]:
> >> >> Partition [
> >> >> > > > > > > > EventServiceUpsertTopic,13] completed preferred replica
> >> >> leader
> >> >> > > > > > election.
> >> >> > > > > > > > New leader is 1
> >> >> > > > > > > > 4. Broker 1 is still behind, but it sets the high water
> >> >> mark to
> >> >> > > > > 971!!!
> >> >> > > > > > > >
> >> >> > > > > > > > Broker 1 Log:
> >> >> > > > > > > >
> >> >> > > > > > > > [2014-07-24 14:44:26,999] [INFO ]
> >> >> [kafka-request-handler-6]
> >> >> > > > > > > > [kafka.server.ReplicaFetcherManager]
> >> >> [ReplicaFetcherManager on
> >> >> > > > > broker
> >> >> > > > > > 1]
> >> >> > > > > > > > Removed fetcher for partitions
> >> [EventServiceUpsertTopic,13]
> >> >> > > > > > > >
> >> >> > > > > > > > [2014-07-24 14:44:27,000] [DEBUG]
> >> >> [kafka-request-handler-6]
> >> >> > > > > > > > [kafka.cluster.Partition] Partition
> >> >> > [EventServiceUpsertTopic,13]
> >> >> > > > on
> >> >> > > > > > > broker
> >> >> > > > > > > > 1: Old hw for partition [EventServiceUpsertTopic,13] is
> >> 970.
> >> >> > New
> >> >> > > hw
> >> >> > > > > is
> >> >> > > > > > > -1.
> >> >> > > > > > > > All leo's are -1,971
> >> >> > > > > > > >
> >> >> > > > > > > > [2014-07-24 14:44:27,098] [DEBUG]
> >> >> [kafka-request-handler-3]
> >> >> > > > > > > > [kafka.server.KafkaApis] [KafkaApi-1] Maybe update
> >> >> partition
> >> >> > HW
> >> >> > > > due
> >> >> > > > > to
> >> >> > > > > > > > fetch request: Name: FetchRequest; Version: 0;
> >> >> CorrelationId:
> >> >> > 1;
> >> >> > > > > > > ClientId:
> >> >> > > > > > > > ReplicaFetcherThread-0-1; ReplicaId: 0; MaxWait: 500
> ms;
> >> >> > > MinBytes:
> >> >> > > > 1
> >> >> > > > > > > bytes;
> >> >> > > > > > > > RequestInfo: [EventServiceUpsertTopic,13] ->
> >> >> > > > > > > > PartitionFetchInfo(971,1048576), <Snipped>
> >> >> > > > > > > >
> >> >> > > > > > > > [2014-07-24 14:44:27,098] [DEBUG]
> >> >> [kafka-request-handler-3]
> >> >> > > > > > > > [kafka.cluster.Partition] Partition
> >> >> > [EventServiceUpsertTopic,13]
> >> >> > > > on
> >> >> > > > > > > broker
> >> >> > > > > > > > 1: Recording follower 0 position 971 for partition [
> >> >> > > > > > > > EventServiceUpsertTopic,13].
> >> >> > > > > > > >
> >> >> > > > > > > > [2014-07-24 14:44:27,100] [DEBUG]
> >> >> [kafka-request-handler-3]
> >> >> > > > > > > > [kafka.cluster.Partition] Partition
> >> >> > [EventServiceUpsertTopic,13]
> >> >> > > > on
> >> >> > > > > > > broker
> >> >> > > > > > > > 1: Highwatermark for partition
> >> [EventServiceUpsertTopic,13]
> >> >> > > updated
> >> >> > > > > to
> >> >> > > > > > > 971
> >> >> > > > > > > > 5. Consumer is none the wiser. All data that was in
> >> offsets
> >> >> > > 972-975
> >> >> > > > > > > doesn't
> >> >> > > > > > > > show up!
> >> >> > > > > > > >
> >> >> > > > > > > > I tried this with 2 initial replicas, and adding a 3rd
> >> >> which is
> >> >> > > > > > supposed
> >> >> > > > > > > to
> >> >> > > > > > > > be the leader for some new partitions, and this problem
> >> also
> >> >> > > > happens
> >> >> > > > > > > there.
> >> >> > > > > > > > The log on the old leader gets truncated to the offset
> on
> >> >> the
> >> >> > new
> >> >> > > > > > leader.
> >> >> > > > > > > > What's the solution? Can I make a new broker leader for
> >> >> > > partitions
> >> >> > > > > that
> >> >> > > > > > > are
> >> >> > > > > > > > currently active without losing data?
> >> >> > > > > > > >
> >> >> > > > > > > > Thanks,
> >> >> > > > > > > > Jad.
> >> >> > > > > > > >
> >> >> > > > > > > > --
> >> >> > > > > > > > *Jad Naous* | Engineering | AppDynamics
> >> >> > > > > > > > <http://www.appdynamics.com>
> >> >> > > > > > > >
> >> >> > > > > > >
> >> >> > > > > > >
> >> >> > > > > > >
> >> >> > > > > > > --
> >> >> > > > > > > -- Guozhang
> >> >> > > > > > >
> >> >> > > > > >
> >> >> > > > > >
> >> >> > > > > >
> >> >> > > > > > --
> >> >> > > > > > *Jad Naous* | Engineering | AppDynamics
> >> >> > > > > > <http://www.appdynamics.com>
> >> >> > > > > >
> >> >> > > > >
> >> >> > > > >
> >> >> > > > >
> >> >> > > > > --
> >> >> > > > > -- Guozhang
> >> >> > > > >
> >> >> > > >
> >> >> > >
> >> >> >
> >> >> >
> >> >> >
> >> >> > --
> >> >> > -- Guozhang
> >> >> >
> >> >>
> >> >>
> >> >>
> >> >> --
> >> >> *Jad Naous* | Engineering | AppDynamics
> >> >> <http://www.appdynamics.com>
> >> >>
> >> >
> >> >
> >> >
> >> > --
> >> > -- Guozhang
> >> >
> >>
> >>
> >>
> >> --
> >> -- Guozhang
> >>
> >
> >
> >
> > --
> > *Jad Naous* | Engineering | AppDynamics
> > <http://www.appdynamics.com>
> >
>
>
>
> --
> -- Guozhang
>
--
*Jad Naous* | Engineering | AppDynamics
<http://www.appdynamics.com>
