>> So, I am guessing now that the 499 bytes written message was going to a buffered socket channel, that received the bytes, but didn't actually send them out at the os level. No exception is thrown, so I am wondering whether this socket is ever flushed, or if it just quietly fails to send its data. Thoughts?
That's correct. When the producer sends some data, it enters the os socket buffer on the producer. Once it is flushed, it is then buffered in the server's socket buffer before being read by the server and appended to the log. On producer or server shutdown, messages can be lost since they are not flushed from the kernel's socket buffer. In 0.7, you will not notice any error. In 0.8, your producer will wait to read the response from the server and either timeout or throw a broken pipe exception. In this case, you can choose to retry. I think you are referring to yet another issue that only happens on the 0.7 zookeeper based producer. If you shutdown any broker, messages in the async producer's queue are lost. This is because, it load balances and picks a target broker when the message enters the queue. So even if it can detect the broker is gone, it cannot change the target broker for the messages in the producer's queue. Both of these issues are fixed in 0.8. Thanks, Neha On Thu, Mar 28, 2013 at 11:44 AM, Jason Rosenberg <[email protected]> wrote: > Ok, sorry, > > I see now that in fact, it does close all logs during LogManager.close(), > which deeper in the code flushes logSegments. So it doesn't do so as > explicitly as in LogManager.flushAllLogs during the normally scheduled > flush interval. > > The confusing thing, is that I clearly see my message being sent from the > producer, e.g.: > > 2013-03-28 10:40:39,090 TRACE [ProducerSendThread-1585330284] > network.BoundedByteBufferSend - 499 bytes written. > > Here's the shutdown sequence on the server, which appears to complete > before the last message was sent above: > > 2013-03-28 10:40:38,660 INFO [Thread-27] server.KafkaServer - Shutting > down Kafka server > 2013-03-28 10:40:38,661 INFO [Thread-27] utils.KafkaScheduler - shutdown > scheduler kafka-logcleaner- > 2013-03-28 10:40:38,670 INFO [Thread-27] utils.KafkaScheduler - shutdown > scheduler kafka-logflusher- > 2013-03-28 10:40:38,673 DEBUG [Thread-27] message.FileMessageSet - flush > time 2 > 2013-03-28 10:40:38,673 DEBUG [Thread-27] message.FileMessageSet - flush > high water mark:227489 > 2013-03-28 10:40:38,674 DEBUG [Thread-27] message.FileMessageSet - flush > time 0 > 2013-03-28 10:40:38,674 DEBUG [Thread-27] message.FileMessageSet - flush > high water mark:37271451 > ... > ... > 2013-03-28 10:40:38,683 INFO [Thread-27] server.KafkaServer - Kafka server > shut down completed > > So, I am guessing now that the 499 bytes written message was going to a > buffered socket channel, that received the bytes, but didn't actually send > them out at the os level. No exception is thrown, so I am wondering > whether this socket is ever flushed, or if it just quietly fails to send > its data. Thoughts? Even if I switched to use a broker list (instead of > zk for broker discovery), it seems that if no exception is thrown on the > send to a closed broker, there's no way to manage retries, etc. > > > Looking deeper, it looks like after I call send on my async producer, there > are log messages coming from zk indicating the broker list has changed, but > it's too late for the async producer to reroute the message: > > 2013-03-28 10:40:38,440 DEBUG [...] producer.Producer - Sending message to > broker .... > 2013-03-28 10:40:38,440 DEBUG [...] producer.ProducerPool - Fetching async > producer for broker id: ... > 2013-03-28 10:40:38,440 DEBUG [...] producer.ProducerPool - Sending message > <broker shuts down> > 2013-03-28 10:40:38,694 DEBUG [ZkClient-EventThread-35-localhost:26101] > producer.ZKBrokerPartitionInfo$BrokerTopicsListener - > [BrokerTopicsListener] List of brokers changed at /brokers/topics/... > ... > ... > 2013-03-28 10:40:39,089 DEBUG [ProducerSendThread-1585330284] > async.ProducerSendThread - 1001 ms elapsed. Queue time reached. Sending.. > 2013-03-28 10:40:39,089 DEBUG [ProducerSendThread-1585330284] > async.ProducerSendThread - Handling 1 events > 2013-03-28 10:40:39,089 TRACE [ProducerSendThread-1585330284] > async.DefaultEventHandler - Handling event for Topic: ... > 2013-03-28 10:40:39,090 TRACE [ProducerSendThread-1585330284] > async.DefaultEventHandler - Sending 1 messages with no compression to topic > ... > <send still goes to previously shut down broker> > 2013-03-28 10:40:39,090 TRACE [ProducerSendThread-1585330284] > network.BoundedByteBufferSend - 499 bytes written. > > I saw issue: https://issues.apache.org/jira/browse/KAFKA-253 > > It seems that this could be relevant. I'm not sure was that fix ever > applied to 0.7.x? > > Will these issues go away with 0.8? > > Jason > > > On Thu, Mar 28, 2013 at 8:31 AM, Jun Rao <[email protected]> wrote: >> Jason, >> >> Our shutdown sequence in 0.7 is the following in KafkaServer.shutdown(): >> >> 1. close socket channel >> 2. close log manager >> 2.1 for each log >> 2.2 flush last segment and close file channel >> >> So, it does what you are suggesting. All unflushed data are supposed to be >> written to disk on a clean shutdown. >> >> Thanks, >> >> Jun >> >> >> On Thu, Mar 28, 2013 at 8:10 AM, Jason Rosenberg <[email protected]> wrote: >> >>> Hi Neha, >>> >>> I enabled TRACE logging on the producer, and verified that it >>> successfully wrote out the bytes to the server (but this was after the >>> last log flush on the server, where trace logging was also enabled). >>> >>> I'm quite certain that the server is receiving the lost data, and then >>> shutting down, before flushing the last few log lines received. >>> >>> It would seem to be an easy fix to update KafkaServer.shutdown, to >>> close all sockets, then do one final log flush that doesn't wait for >>> the flush interval, and then close down. >>> >>> Jason >>> >>> On Thu, Mar 28, 2013 at 6:45 AM, Neha Narkhede <[email protected]> >>> wrote: >>> > Jason, >>> > >>> > How do you know the server had written the lost data to its log ? In >>> Kafka >>> > 0.7, data could be lost from the producer's or server's socket buffer. >>> You >>> > can verify this by running DumpLogSegments before and after shutdown. >>> > >>> > Thanks, >>> > Neha >>> > >>> > On Thursday, March 28, 2013, SuoNayi wrote: >>> > >>> >> Hi, the mail list has been moved to >>> >> [email protected]<javascript:;>long > time ago. >>> >> You may resubscribe the list or the list may be removed in future. >>> >> >>> >> >>> >> >>> >> >>> >> At 2013-03-28 15:39:07,"Jason Rosenberg" <[email protected] > <javascript:;>> >>> >> wrote: >>> >> >Hi, >>> >> > >>> >> >We are managing our kafka clusters by doing rolling restarts (e.g. we >>> >> >cleanly shutdown and restart each broker one at a time). >>> >> > >>> >> >I'm working through an issue, whereby we are losing messages sent to > a >>> >> >kafka broker, right before the broker is shutdown cleanly. >>> >> > >>> >> >I'm still using 0.7.2, so not sure if this is also the behavior in > 0.8. >>> >> > >>> >> >I am trying to cleanly shutting down the kafka server by calling >>> >> >kafka.server.KafkaServer.shutdown(). I then wait for the shutdown to >>> >> >complete, by calling awaitShutdown(). >>> >> > >>> >> >However, it looks like this does not attempt to flush all logs before >>> >> >quitting, as I had assumed was the case. Is there a way to make sure >>> >> >this happens before the server stops? >>> >> > >>> >> >Jason >>> >> >>>
