old_contents: #!/usr/bin/env python
  import threading, logging, time, collections
  
  from kafka.client import KafkaClient
  from kafka.consumer import SimpleConsumer
  from kafka.producer import SimpleProducer
  
  msg_size = 524288
  
  class Producer(threading.Thread):
      daemon = True
      big_msg = "1" * msg_size
  
      def run(self):
          client = KafkaClient("localhost", 9092)
          producer = SimpleProducer(client)
          self.sent = 0
  
          while True:
              producer.send_messages('my-topic', self.big_msg)
              self.sent += 1
  
  
  class Consumer(threading.Thread):
      daemon = True
  
      def run(self):
          client = KafkaClient("localhost", 9092)
          consumer = SimpleConsumer(client, "test-group", "my-topic",
              max_buffer_size = None,
          )
          self.valid = 0
          self.invalid = 0
  
          for message in consumer:
              if len(message.message.value) == msg_size:
                  self.valid += 1
              else:
                  self.invalid += 1
  
  def main():
      threads = [
          Producer(),
          Consumer()
      ]
  
      for t in threads:
          t.start()
  
      time.sleep(10)
      print 'Messages sent: %d' % threads[0].sent
      print 'Messages recvd: %d' % threads[1].valid
      print 'Messages invalid: %d' % threads[1].invalid
  
  if __name__ == "__main__":
      logging.basicConfig(
          format='%(asctime)s.%(msecs)s:%(name)s:%(thread)d:%(levelname)s:%(process)d:%(message)s',
          level=logging.DEBUG
          )
      main()
  
new_contents: #!/usr/bin/env python
  import threading, logging, time, collections
  
  from kafka.client import KafkaClient
  from kafka.consumer import SimpleConsumer
  from kafka.producer import SimpleProducer
  
  msg_size = 524288
  
  class Producer(threading.Thread):
      daemon = True
      big_msg = "1" * msg_size
  
      def run(self):
          client = KafkaClient("localhost:9092")
          producer = SimpleProducer(client)
          self.sent = 0
  
          while True:
              producer.send_messages('my-topic', self.big_msg)
              self.sent += 1
  
  
  class Consumer(threading.Thread):
      daemon = True
  
      def run(self):
          client = KafkaClient("localhost:9092")
          consumer = SimpleConsumer(client, "test-group", "my-topic",
              max_buffer_size = None,
          )
          self.valid = 0
          self.invalid = 0
  
          for message in consumer:
              if len(message.message.value) == msg_size:
                  self.valid += 1
              else:
                  self.invalid += 1
  
  def main():
      threads = [
          Producer(),
          Consumer()
      ]
  
      for t in threads:
          t.start()
  
      time.sleep(10)
      print 'Messages sent: %d' % threads[0].sent
      print 'Messages recvd: %d' % threads[1].valid
      print 'Messages invalid: %d' % threads[1].invalid
  
  if __name__ == "__main__":
      logging.basicConfig(
          format='%(asctime)s.%(msecs)s:%(name)s:%(thread)d:%(levelname)s:%(process)d:%(message)s',
          level=logging.DEBUG
          )
      main()
  
subject: Update load_example