Extra - Kafka Load.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 207,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import os\n",
    "import ipyparallel as ipp\n",
    "\n",
    "rc = ipp.Client()\n",
    "ar = rc[:].apply_async(os.getpid)\n",
    "pid_map = ar.get_dict()\n",
    "v = rc[:]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 208,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "docker_ip = '192.168.99.100'\n",
    "client = pykafka.KafkaClient(hosts=\"{}:9092\".format(docker_ip))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 137,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[None, None, None, None]"
      ]
     },
     "execution_count": 137,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "v.scatter('keys', range(10000000)).get()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 209,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "keys = [0,1]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 210,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 2.58 ms, sys: 4.9 ms, total: 7.48 ms\n",
      "Wall time: 33.7 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "#%%px \n",
    "import pykafka\n",
    " \n",
    "docker_ip = '192.168.99.100'\n",
    "client = pykafka.KafkaClient(hosts=\"{}:9092\".format(docker_ip))\n",
    "t = client.topics['input']\n",
    "producer = t.get_producer()\n",
    "producer.produce([('foo: {}'.format(i), i) for i in keys])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 212,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "c = client.topics['input'].get_simple_consumer()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 215,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "m = c.consume(block=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 216,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "('1', 'foo: 1', 1)"
      ]
     },
     "execution_count": 216,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING:kazoo.client:Connection dropped: socket connection broken\n"
     ]
    }
   ],
   "source": [
    "m.value, m.partition_key, m.offset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 99,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING:kazoo.client:Connection dropped: outstanding heartbeat ping not received\n",
      "WARNING:kazoo.client:Transition to CONNECTING\n",
      "WARNING:kazoo.client:Connection dropped: outstanding heartbeat ping not received\n",
      "WARNING:kazoo.client:Transition to CONNECTING\n",
      "WARNING:kazoo.client:Connection dropped: outstanding heartbeat ping not received\n",
      "WARNING:kazoo.client:Transition to CONNECTING\n",
      "WARNING:kazoo.client:Connection dropped: outstanding heartbeat ping not received\n",
      "WARNING:kazoo.client:Transition to CONNECTING\n",
      "WARNING:kazoo.client:Connection dropped: outstanding heartbeat ping not received\n",
      "WARNING:kazoo.client:Transition to CONNECTING\n",
      "WARNING:kazoo.client:Connection dropped: outstanding heartbeat ping not received\n",
      "WARNING:kazoo.client:Transition to CONNECTING\n",
      "WARNING:kazoo.client:Connection dropped: outstanding heartbeat ping not received\n",
      "WARNING:kazoo.client:Transition to CONNECTING\n",
      "WARNING:kazoo.client:Connection dropped: outstanding heartbeat ping not received\n",
      "WARNING:kazoo.client:Transition to CONNECTING\n",
      "WARNING:kazoo.client:Connection dropped: outstanding heartbeat ping not received\n",
      "WARNING:kazoo.client:Transition to CONNECTING\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "Exception in thread Thread-66:\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 810, in __bootstrap_inner\n",
      "    self.run()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 763, in run\n",
      "    self.__target(*self.__args, **self.__kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 200, in checker\n",
      "    self._check_held_partitions()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 467, in _check_held_partitions\n",
      "    all_partitions = self._zookeeper.get_children(self._topic_path)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/client.py\", line 1078, in get_children\n",
      "    return self.get_children_async(path, watch, include_data).get()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/handlers/utils.py\", line 72, in get\n",
      "    raise self._exception\n",
      "SessionExpiredError\n",
      "\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Connection dropped: socket connection error: None\n",
      "WARNING:kazoo.client:Session has expired\n",
      "WARNING:kazoo.client:Session has expired\n",
      "WARNING:kazoo.client:Session has expired\n",
      "WARNING:kazoo.client:Session has expired\n",
      "WARNING:kazoo.client:Session has expired\n",
      "WARNING:kazoo.client:Session has expired\n",
      "WARNING:kazoo.client:Session has expired\n",
      "WARNING:kazoo.client:Session has expired\n",
      "Exception in thread Thread-90:\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 810, in __bootstrap_inner\n",
      "    self.run()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 763, in run\n",
      "    self.__target(*self.__args, **self.__kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/simpleconsumer.py\", line 255, in autocommitter\n",
      "    self._auto_commit()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/simpleconsumer.py\", line 318, in _auto_commit\n",
      "    self.commit_offsets()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/simpleconsumer.py\", line 338, in commit_offsets\n",
      "    self._consumer_group, 1, 'pykafka', reqs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/broker.py\", line 309, in commit_consumer_group_offsets\n",
      "    return self._offsets_channel_req_handler.request(req).get(OffsetCommitResponse)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/handlers.py\", line 55, in get\n",
      "    raise self.error\n",
      "timeout: timed out\n",
      "\n",
      "Exception in thread Thread-88:\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 810, in __bootstrap_inner\n",
      "    self.run()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 763, in run\n",
      "    self.__target(*self.__args, **self.__kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/simpleconsumer.py\", line 255, in autocommitter\n",
      "    self._auto_commit()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/simpleconsumer.py\", line 318, in _auto_commit\n",
      "    self.commit_offsets()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/simpleconsumer.py\", line 338, in commit_offsets\n",
      "    self._consumer_group, 1, 'pykafka', reqs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/broker.py\", line 309, in commit_consumer_group_offsets\n",
      "    return self._offsets_channel_req_handler.request(req).get(OffsetCommitResponse)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/handlers.py\", line 55, in get\n",
      "    raise self.error\n",
      "timeout: timed out\n",
      "\n",
      "WARNING:kazoo.client:Session has expired\n",
      "Exception in thread Thread-111:\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 810, in __bootstrap_inner\n",
      "    self.run()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 763, in run\n",
      "    self.__target(*self.__args, **self.__kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 36, in wrapper\n",
      "    return func(*args, **kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 319, in _get_children\n",
      "    self._path, self._watcher)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/client.py\", line 272, in _retry\n",
      "    return self._retry.copy()(*args, **kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/retry.py\", line 123, in __call__\n",
      "    return func(*args, **kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/client.py\", line 1078, in get_children\n",
      "    return self.get_children_async(path, watch, include_data).get()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/handlers/utils.py\", line 78, in get\n",
      "    raise self._exception\n",
      "NoNodeError: ((), {})\n",
      "\n",
      "Exception in thread Thread-112:\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 810, in __bootstrap_inner\n",
      "    self.run()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 763, in run\n",
      "    self.__target(*self.__args, **self.__kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 36, in wrapper\n",
      "    return func(*args, **kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 319, in _get_children\n",
      "    self._path, self._watcher)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/client.py\", line 272, in _retry\n",
      "    return self._retry.copy()(*args, **kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/retry.py\", line 123, in __call__\n",
      "    return func(*args, **kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/client.py\", line 1078, in get_children\n",
      "    return self.get_children_async(path, watch, include_data).get()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/handlers/utils.py\", line 78, in get\n",
      "    raise self._exception\n",
      "NoNodeError: ((), {})\n",
      "\n",
      "ERROR:kazoo.recipe.watchers:'JDENNISON-MAC.local:791b4536-eff9-4c98-9da7-de8e4018ae3b' is not in list\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 333, in _get_children\n",
      "    result = self._func(children)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 492, in _consumers_changed\n",
      "    self._rebalance()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 397, in _rebalance\n",
      "    partitions = self._decide_partitions(participants)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 293, in _decide_partitions\n",
      "    idx = participants.index(self._consumer_id)\n",
      "ValueError: 'JDENNISON-MAC.local:791b4536-eff9-4c98-9da7-de8e4018ae3b' is not in list\n",
      "Exception in thread Thread-108:\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 810, in __bootstrap_inner\n",
      "    self.run()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 763, in run\n",
      "    self.__target(*self.__args, **self.__kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 36, in wrapper\n",
      "    return func(*args, **kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 333, in _get_children\n",
      "    result = self._func(children)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 492, in _consumers_changed\n",
      "    self._rebalance()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 397, in _rebalance\n",
      "    partitions = self._decide_partitions(participants)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 293, in _decide_partitions\n",
      "    idx = participants.index(self._consumer_id)\n",
      "ValueError: 'JDENNISON-MAC.local:791b4536-eff9-4c98-9da7-de8e4018ae3b' is not in list\n",
      "\n",
      "Exception in thread Thread-94:\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 810, in __bootstrap_inner\n",
      "    self.run()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 763, in run\n",
      "    self.__target(*self.__args, **self.__kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/simpleconsumer.py\", line 255, in autocommitter\n",
      "    self._auto_commit()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/simpleconsumer.py\", line 318, in _auto_commit\n",
      "    self.commit_offsets()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/simpleconsumer.py\", line 342, in commit_offsets\n",
      "    partitions_by_id=self._partitions_by_id)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/utils/error_handlers.py\", line 50, in handle_partition_responses\n",
      "    owned_partition = partitions_by_id[partition_id]\n",
      "KeyError: 6\n",
      "\n",
      "Exception in thread Thread-96:\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 810, in __bootstrap_inner\n",
      "    self.run()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 763, in run\n",
      "    self.__target(*self.__args, **self.__kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/simpleconsumer.py\", line 255, in autocommitter\n",
      "    self._auto_commit()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/simpleconsumer.py\", line 318, in _auto_commit\n",
      "    self.commit_offsets()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/simpleconsumer.py\", line 342, in commit_offsets\n",
      "    partitions_by_id=self._partitions_by_id)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/utils/error_handlers.py\", line 50, in handle_partition_responses\n",
      "    owned_partition = partitions_by_id[partition_id]\n",
      "KeyError: 3\n",
      "\n",
      "ERROR:kazoo.recipe.watchers:'JDENNISON-MAC.local:61bf886f-e3fc-46d5-aa91-c54309d0f079' is not in list\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 333, in _get_children\n",
      "    result = self._func(children)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 492, in _consumers_changed\n",
      "    self._rebalance()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 397, in _rebalance\n",
      "    partitions = self._decide_partitions(participants)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 293, in _decide_partitions\n",
      "    idx = participants.index(self._consumer_id)\n",
      "ValueError: 'JDENNISON-MAC.local:61bf886f-e3fc-46d5-aa91-c54309d0f079' is not in list\n",
      "Exception in thread Thread-100:\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 810, in __bootstrap_inner\n",
      "    self.run()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 763, in run\n",
      "    self.__target(*self.__args, **self.__kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 36, in wrapper\n",
      "    return func(*args, **kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 333, in _get_children\n",
      "    result = self._func(children)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 492, in _consumers_changed\n",
      "    self._rebalance()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 397, in _rebalance\n",
      "    partitions = self._decide_partitions(participants)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 293, in _decide_partitions\n",
      "    idx = participants.index(self._consumer_id)\n",
      "ValueError: 'JDENNISON-MAC.local:61bf886f-e3fc-46d5-aa91-c54309d0f079' is not in list\n",
      "\n",
      "ERROR:kazoo.recipe.watchers:0\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 333, in _get_children\n",
      "    result = self._func(children)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 492, in _consumers_changed\n",
      "    self._rebalance()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 386, in _rebalance\n",
      "    self.commit_offsets()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 521, in commit_offsets\n",
      "    return self._consumer.commit_offsets()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/simpleconsumer.py\", line 342, in commit_offsets\n",
      "    partitions_by_id=self._partitions_by_id)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/utils/error_handlers.py\", line 50, in handle_partition_responses\n",
      "    owned_partition = partitions_by_id[partition_id]\n",
      "KeyError: 0\n",
      "Exception in thread Thread-104:\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 810, in __bootstrap_inner\n",
      "    self.run()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 763, in run\n",
      "    self.__target(*self.__args, **self.__kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 36, in wrapper\n",
      "    return func(*args, **kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 333, in _get_children\n",
      "    result = self._func(children)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 492, in _consumers_changed\n",
      "    self._rebalance()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 386, in _rebalance\n",
      "    self.commit_offsets()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 521, in commit_offsets\n",
      "    return self._consumer.commit_offsets()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/simpleconsumer.py\", line 342, in commit_offsets\n",
      "    partitions_by_id=self._partitions_by_id)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/utils/error_handlers.py\", line 50, in handle_partition_responses\n",
      "    owned_partition = partitions_by_id[partition_id]\n",
      "KeyError: 0\n",
      "\n",
      "ERROR:kazoo.recipe.watchers:3\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 333, in _get_children\n",
      "    result = self._func(children)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 492, in _consumers_changed\n",
      "    self._rebalance()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 386, in _rebalance\n",
      "    self.commit_offsets()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 521, in commit_offsets\n",
      "    return self._consumer.commit_offsets()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/simpleconsumer.py\", line 342, in commit_offsets\n",
      "    partitions_by_id=self._partitions_by_id)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/utils/error_handlers.py\", line 50, in handle_partition_responses\n",
      "    owned_partition = partitions_by_id[partition_id]\n",
      "KeyError: 3\n",
      "Exception in thread Thread-107:\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 810, in __bootstrap_inner\n",
      "    self.run()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 763, in run\n",
      "    self.__target(*self.__args, **self.__kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 36, in wrapper\n",
      "    return func(*args, **kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 333, in _get_children\n",
      "    result = self._func(children)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 492, in _consumers_changed\n",
      "    self._rebalance()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 386, in _rebalance\n",
      "    self.commit_offsets()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 521, in commit_offsets\n",
      "    return self._consumer.commit_offsets()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/simpleconsumer.py\", line 342, in commit_offsets\n",
      "    partitions_by_id=self._partitions_by_id)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/utils/error_handlers.py\", line 50, in handle_partition_responses\n",
      "    owned_partition = partitions_by_id[partition_id]\n",
      "KeyError: 3\n",
      "\n",
      "ERROR:kazoo.recipe.watchers:'JDENNISON-MAC.local:abc990bd-8119-4579-baa4-3e8095e3a1bd' is not in list\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 333, in _get_children\n",
      "    result = self._func(children)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 492, in _consumers_changed\n",
      "    self._rebalance()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 397, in _rebalance\n",
      "    partitions = self._decide_partitions(participants)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 293, in _decide_partitions\n",
      "    idx = participants.index(self._consumer_id)\n",
      "ValueError: 'JDENNISON-MAC.local:abc990bd-8119-4579-baa4-3e8095e3a1bd' is not in list\n",
      "Exception in thread Thread-114:\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 810, in __bootstrap_inner\n",
      "    self.run()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 763, in run\n",
      "    self.__target(*self.__args, **self.__kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 36, in wrapper\n",
      "    return func(*args, **kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 333, in _get_children\n",
      "    result = self._func(children)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 492, in _consumers_changed\n",
      "    self._rebalance()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 397, in _rebalance\n",
      "    partitions = self._decide_partitions(participants)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 293, in _decide_partitions\n",
      "    idx = participants.index(self._consumer_id)\n",
      "ValueError: 'JDENNISON-MAC.local:abc990bd-8119-4579-baa4-3e8095e3a1bd' is not in list\n",
      "\n",
      "WARNING:pykafka.balancedconsumer:Internal partition registry doesn't match ZooKeeper!\n",
      "Exception in thread Thread-42:\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 810, in __bootstrap_inner\n",
      "    self.run()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 763, in run\n",
      "    self.__target(*self.__args, **self.__kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 200, in checker\n",
      "    self._check_held_partitions()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 480, in _check_held_partitions\n",
      "    self._rebalance()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 397, in _rebalance\n",
      "    partitions = self._decide_partitions(participants)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 293, in _decide_partitions\n",
      "    idx = participants.index(self._consumer_id)\n",
      "ValueError: 'JDENNISON-MAC.local:abc990bd-8119-4579-baa4-3e8095e3a1bd' is not in list\n",
      "\n",
      "WARNING:pykafka.balancedconsumer:Internal partition registry doesn't match ZooKeeper!\n",
      "Exception in thread Thread-76:\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 810, in __bootstrap_inner\n",
      "    self.run()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 763, in run\n",
      "    self.__target(*self.__args, **self.__kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 200, in checker\n",
      "    self._check_held_partitions()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 480, in _check_held_partitions\n",
      "    self._rebalance()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 386, in _rebalance\n",
      "    self.commit_offsets()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 521, in commit_offsets\n",
      "    return self._consumer.commit_offsets()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/simpleconsumer.py\", line 342, in commit_offsets\n",
      "    partitions_by_id=self._partitions_by_id)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/utils/error_handlers.py\", line 50, in handle_partition_responses\n",
      "    owned_partition = partitions_by_id[partition_id]\n",
      "KeyError: 6\n",
      "\n",
      "ERROR:kazoo.recipe.watchers:3\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 333, in _get_children\n",
      "    result = self._func(children)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 492, in _consumers_changed\n",
      "    self._rebalance()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 386, in _rebalance\n",
      "    self.commit_offsets()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 521, in commit_offsets\n",
      "    return self._consumer.commit_offsets()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/simpleconsumer.py\", line 342, in commit_offsets\n",
      "    partitions_by_id=self._partitions_by_id)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/utils/error_handlers.py\", line 50, in handle_partition_responses\n",
      "    owned_partition = partitions_by_id[partition_id]\n",
      "KeyError: 3\n",
      "ERROR:kazoo.handlers.threading:Exception in worker queue thread\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/handlers/threading.py\", line 99, in _thread_worker\n",
      "    func()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/handlers/threading.py\", line 196, in <lambda>\n",
      "    self.callback_queue.put(lambda: callback.func(*callback.args))\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 341, in _watcher\n",
      "    self._get_children(event)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 36, in wrapper\n",
      "    return func(*args, **kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 333, in _get_children\n",
      "    result = self._func(children)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 492, in _consumers_changed\n",
      "    self._rebalance()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 386, in _rebalance\n",
      "    self.commit_offsets()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 521, in commit_offsets\n",
      "    return self._consumer.commit_offsets()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/simpleconsumer.py\", line 342, in commit_offsets\n",
      "    partitions_by_id=self._partitions_by_id)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/utils/error_handlers.py\", line 50, in handle_partition_responses\n",
      "    owned_partition = partitions_by_id[partition_id]\n",
      "KeyError: 3\n",
      "ERROR:kazoo.recipe.watchers:'JDENNISON-MAC.local:791b4536-eff9-4c98-9da7-de8e4018ae3b' is not in list\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 333, in _get_children\n",
      "    result = self._func(children)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 492, in _consumers_changed\n",
      "    self._rebalance()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 397, in _rebalance\n",
      "    partitions = self._decide_partitions(participants)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 293, in _decide_partitions\n",
      "    idx = participants.index(self._consumer_id)\n",
      "ValueError: 'JDENNISON-MAC.local:791b4536-eff9-4c98-9da7-de8e4018ae3b' is not in list\n",
      "ERROR:kazoo.handlers.threading:Exception in worker queue thread\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/handlers/threading.py\", line 99, in _thread_worker\n",
      "    func()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/handlers/threading.py\", line 196, in <lambda>\n",
      "    self.callback_queue.put(lambda: callback.func(*callback.args))\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 341, in _watcher\n",
      "    self._get_children(event)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 36, in wrapper\n",
      "    return func(*args, **kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/kazoo/recipe/watchers.py\", line 333, in _get_children\n",
      "    result = self._func(children)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 492, in _consumers_changed\n",
      "    self._rebalance()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 397, in _rebalance\n",
      "    partitions = self._decide_partitions(participants)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 293, in _decide_partitions\n",
      "    idx = participants.index(self._consumer_id)\n",
      "ValueError: 'JDENNISON-MAC.local:791b4536-eff9-4c98-9da7-de8e4018ae3b' is not in list\n",
      "WARNING:pykafka.balancedconsumer:Internal partition registry doesn't match ZooKeeper!\n",
      "Exception in thread Thread-26:\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 810, in __bootstrap_inner\n",
      "    self.run()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.py\", line 763, in run\n",
      "    self.__target(*self.__args, **self.__kwargs)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 200, in checker\n",
      "    self._check_held_partitions()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 480, in _check_held_partitions\n",
      "    self._rebalance()\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 397, in _rebalance\n",
      "    partitions = self._decide_partitions(participants)\n",
      "  File \"/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 293, in _decide_partitions\n",
      "    idx = participants.index(self._consumer_id)\n",
      "ValueError: 'JDENNISON-MAC.local:791b4536-eff9-4c98-9da7-de8e4018ae3b' is not in list\n",
      "\n"
     ]
    },
    {
     "ename": "KeyboardInterrupt",
     "evalue": "",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mKeyboardInterrupt\u001b[0m                         Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-99-92202603baae>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m()\u001b[0m\n\u001b[0;32m----> 1\u001b[0;31m \u001b[0mget_ipython\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mrun_cell_magic\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34mu'time'\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34mu''\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34mu'%%px\\nimport pykafka\\nimport time\\ndocker_ip = \\'192.168.99.100\\'\\nclient = pykafka.KafkaClient(hosts=\"{}:9092\".format(docker_ip))\\ninput_topic = client.topics[\\'input\\']\\noutput_topic = client.topics[\\'output\\']\\nproducer = output_topic.get_producer()\\n\\nbalanced_consumer = input_topic.get_balanced_consumer(\\n    consumer_group=\\'testgroup4\\',\\n    auto_commit_enable=True,\\n    zookeeper_connect=\\'192.168.99.100:2181\\'\\n)\\ntime.sleep(5)\\n\\nwhile True:\\n    if balanced_consumer:\\n        m = balanced_consumer.consume(block=False)\\n        if not m:\\n            break\\n        producer.produce(\\'out: {}\\'.format(m.value))'\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[0;32m/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/IPython/core/interactiveshell.pyc\u001b[0m in \u001b[0;36mrun_cell_magic\u001b[0;34m(self, magic_name, line, cell)\u001b[0m\n\u001b[1;32m   2291\u001b[0m             \u001b[0mmagic_arg_s\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mvar_expand\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mline\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mstack_depth\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   2292\u001b[0m             \u001b[0;32mwith\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mbuiltin_trap\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 2293\u001b[0;31m                 \u001b[0mresult\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mfn\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mmagic_arg_s\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mcell\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   2294\u001b[0m             \u001b[0;32mreturn\u001b[0m \u001b[0mresult\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   2295\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/IPython/core/magics/execution.pyc\u001b[0m in \u001b[0;36mtime\u001b[0;34m(self, line, cell, local_ns)\u001b[0m\n",
      "\u001b[0;32m/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/IPython/core/magic.pyc\u001b[0m in \u001b[0;36m<lambda>\u001b[0;34m(f, *a, **k)\u001b[0m\n\u001b[1;32m    191\u001b[0m     \u001b[0;31m# but it's overkill for just that one bit of state.\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    192\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0mmagic_deco\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0marg\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 193\u001b[0;31m         \u001b[0mcall\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;32mlambda\u001b[0m \u001b[0mf\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m*\u001b[0m\u001b[0ma\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m**\u001b[0m\u001b[0mk\u001b[0m\u001b[0;34m:\u001b[0m \u001b[0mf\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m*\u001b[0m\u001b[0ma\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m**\u001b[0m\u001b[0mk\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    194\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    195\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mcallable\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0marg\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/IPython/core/magics/execution.pyc\u001b[0m in \u001b[0;36mtime\u001b[0;34m(self, line, cell, local_ns)\u001b[0m\n\u001b[1;32m   1161\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mmode\u001b[0m\u001b[0;34m==\u001b[0m\u001b[0;34m'eval'\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1162\u001b[0m             \u001b[0mst\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mclock2\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1163\u001b[0;31m             \u001b[0mout\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0meval\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mcode\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mglob\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mlocal_ns\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   1164\u001b[0m             \u001b[0mend\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mclock2\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1165\u001b[0m         \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m<timed eval>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m()\u001b[0m\n",
      "\u001b[0;32m/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/IPython/core/interactiveshell.pyc\u001b[0m in \u001b[0;36mrun_cell_magic\u001b[0;34m(self, magic_name, line, cell)\u001b[0m\n\u001b[1;32m   2291\u001b[0m             \u001b[0mmagic_arg_s\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mvar_expand\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mline\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mstack_depth\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   2292\u001b[0m             \u001b[0;32mwith\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mbuiltin_trap\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 2293\u001b[0;31m                 \u001b[0mresult\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mfn\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mmagic_arg_s\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mcell\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   2294\u001b[0m             \u001b[0;32mreturn\u001b[0m \u001b[0mresult\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   2295\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/ipyparallel/client/magics.pyc\u001b[0m in \u001b[0;36mcell_px\u001b[0;34m(self, line, cell)\u001b[0m\n\u001b[1;32m    296\u001b[0m             ar = self.parallel_execute(cell, block=block,\n\u001b[1;32m    297\u001b[0m                                 \u001b[0mgroupby\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0margs\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mgroupby\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 298\u001b[0;31m                                 \u001b[0msave_name\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0margs\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msave_name\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    299\u001b[0m             )\n\u001b[1;32m    300\u001b[0m         \u001b[0;32mfinally\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/ipyparallel/client/magics.pyc\u001b[0m in \u001b[0;36mparallel_execute\u001b[0;34m(self, cell, block, groupby, save_name)\u001b[0m\n\u001b[1;32m    258\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    259\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mblock\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 260\u001b[0;31m             \u001b[0mresult\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mget\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    261\u001b[0m             \u001b[0mresult\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mdisplay_outputs\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mgroupby\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    262\u001b[0m         \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/ipyparallel/client/asyncresult.pyc\u001b[0m in \u001b[0;36mget\u001b[0;34m(self, timeout)\u001b[0m\n\u001b[1;32m     96\u001b[0m         \"\"\"\n\u001b[1;32m     97\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0;32mnot\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mready\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 98\u001b[0;31m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mwait\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mtimeout\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m     99\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    100\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_ready\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/ipyparallel/client/asyncresult.pyc\u001b[0m in \u001b[0;36mwait\u001b[0;34m(self, timeout)\u001b[0m\n\u001b[1;32m    127\u001b[0m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_wait_for_outputs\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mtimeout\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    128\u001b[0m             \u001b[0;32mreturn\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 129\u001b[0;31m         \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_ready\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_client\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mwait\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mmsg_ids\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mtimeout\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    130\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_ready\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    131\u001b[0m             \u001b[0;32mtry\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/ipyparallel/client/client.pyc\u001b[0m in \u001b[0;36mwait\u001b[0;34m(self, jobs, timeout)\u001b[0m\n\u001b[1;32m   1067\u001b[0m             \u001b[0;32mif\u001b[0m \u001b[0mtimeout\u001b[0m \u001b[0;34m>=\u001b[0m \u001b[0;36m0\u001b[0m \u001b[0;32mand\u001b[0m \u001b[0;34m(\u001b[0m \u001b[0mtime\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtime\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m-\u001b[0m\u001b[0mtic\u001b[0m \u001b[0;34m)\u001b[0m \u001b[0;34m>\u001b[0m \u001b[0mtimeout\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1068\u001b[0m                 \u001b[0;32mbreak\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1069\u001b[0;31m             \u001b[0mtime\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msleep\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;36m1e-3\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   1070\u001b[0m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mspin\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1071\u001b[0m         \u001b[0;32mreturn\u001b[0m \u001b[0mlen\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mtheids\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mintersection\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0moutstanding\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m \u001b[0;34m==\u001b[0m \u001b[0;36m0\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mKeyboardInterrupt\u001b[0m: "
     ]
    }
   ],
   "source": [
    "%%time\n",
    "%%px\n",
    "import pykafka\n",
    "import time\n",
    "docker_ip = '192.168.99.100'\n",
    "client = pykafka.KafkaClient(hosts=\"{}:9092\".format(docker_ip))\n",
    "input_topic = client.topics['input']\n",
    "output_topic = client.topics['output']\n",
    "producer = output_topic.get_producer()\n",
    "\n",
    "balanced_consumer = input_topic.get_balanced_consumer(\n",
    "    consumer_group='testgroup4',\n",
    "    auto_commit_enable=True,\n",
    "    zookeeper_connect='192.168.99.100:2181'\n",
    ")\n",
    "time.sleep(5)\n",
    "\n",
    "count = 0\n",
    "while True:\n",
    "    if balanced_consumer:\n",
    "        m = balanced_consumer.consume(block=False)\n",
    "        if not m:\n",
    "            break\n",
    "        c9ount =+ 1\n",
    "        producer.produce(['out: {}'.format(m.value)])\n",
    "        "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 97,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "consumer = client.topics['output'].get_simple_consumer()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 98,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "We counsumed: 0 messages\n",
      "CPU times: user 49 µs, sys: 15 µs, total: 64 µs\n",
      "Wall time: 57.9 µs\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "count = 0\n",
    "\n",
    "while True:\n",
    "    message = consumer.consume(block=False)\n",
    "    if not message:\n",
    "        break\n",
    "    # YOUR CODE HERE\n",
    "    count += 1\n",
    "print \"We counsumed: {} messages\".format(count)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 223,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "input_topic = client.topics['input']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 226,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "balanced_consumer2 = input_topic.get_balanced_consumer(\n",
    "    consumer_group='testgroup65',\n",
    "    auto_commit_enable=True,\n",
    "    zookeeper_connect='192.168.99.100:2181'\n",
    ")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 229,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "ename": "KeyboardInterrupt",
     "evalue": "",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mKeyboardInterrupt\u001b[0m                         Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-229-39f054bcbb22>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m()\u001b[0m\n\u001b[0;32m----> 1\u001b[0;31m \u001b[0mbalanced_consumer2\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mconsume\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[0;32m/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/balancedconsumer.pyc\u001b[0m in \u001b[0;36mconsume\u001b[0;34m(self, block)\u001b[0m\n\u001b[1;32m    507\u001b[0m         \u001b[0;32mwhile\u001b[0m \u001b[0mmessage\u001b[0m \u001b[0;32mis\u001b[0m \u001b[0mNone\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    508\u001b[0m             \u001b[0;32mtry\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 509\u001b[0;31m                 \u001b[0mmessage\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_consumer\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mconsume\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mblock\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mblock\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    510\u001b[0m             \u001b[0;32mexcept\u001b[0m \u001b[0mConsumerStoppedException\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    511\u001b[0m                 \u001b[0;32mcontinue\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/simpleconsumer.pyc\u001b[0m in \u001b[0;36mconsume\u001b[0;34m(self, block)\u001b[0m\n\u001b[1;32m    294\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    295\u001b[0m         \u001b[0;32mwhile\u001b[0m \u001b[0mTrue\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 296\u001b[0;31m             \u001b[0;32mif\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_messages_arrived\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0macquire\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mblocking\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mblock\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mtimeout\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mtimeout\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    297\u001b[0m                 \u001b[0;31m# by passing through this semaphore, we know that at\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    298\u001b[0m                 \u001b[0;31m# least one message is waiting in some queue.\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/Users/jdennison/anaconda/envs/py2/lib/python2.7/site-packages/pykafka/utils/compat.pyc\u001b[0m in \u001b[0;36macquire\u001b[0;34m(self, blocking, timeout)\u001b[0m\n\u001b[1;32m     73\u001b[0m                             \u001b[0;32mif\u001b[0m \u001b[0mtimeout\u001b[0m \u001b[0;34m<=\u001b[0m \u001b[0;36m0\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     74\u001b[0m                                 \u001b[0;32mbreak\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 75\u001b[0;31m                     \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_cond\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mwait\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mtimeout\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m     76\u001b[0m                 \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     77\u001b[0m                     \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_value\u001b[0m \u001b[0;34m-=\u001b[0m \u001b[0;36m1\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/Users/jdennison/anaconda/envs/py2/lib/python2.7/threading.pyc\u001b[0m in \u001b[0;36mwait\u001b[0;34m(self, timeout)\u001b[0m\n\u001b[1;32m    357\u001b[0m                         \u001b[0;32mbreak\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    358\u001b[0m                     \u001b[0mdelay\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mmin\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mdelay\u001b[0m \u001b[0;34m*\u001b[0m \u001b[0;36m2\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mremaining\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;36m.05\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 359\u001b[0;31m                     \u001b[0m_sleep\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mdelay\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    360\u001b[0m                 \u001b[0;32mif\u001b[0m \u001b[0;32mnot\u001b[0m \u001b[0mgotit\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    361\u001b[0m                     \u001b[0;32mif\u001b[0m \u001b[0m__debug__\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mKeyboardInterrupt\u001b[0m: "
     ]
    }
   ],
   "source": [
    "balanced_consumer2.consume()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 83,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "balanced_consumer3 = input_topic.get_balanced_consumer(\n",
    "    consumer_group='testgroup3',\n",
    "    auto_commit_enable=True,\n",
    "    zookeeper_connect='192.168.99.100:2181'\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 84,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{<pykafka.partition.Partition at 0x115d82e90 (id=0)>,\n",
       " <pykafka.partition.Partition at 0x115d82ed0 (id=1)>,\n",
       " <pykafka.partition.Partition at 0x115d82f10 (id=2)>}"
      ]
     },
     "execution_count": 84,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "balanced_consumer1._partitions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 86,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{<pykafka.partition.Partition at 0x115d82f50 (id=3)>,\n",
       " <pykafka.partition.Partition at 0x115d82f90 (id=4)>,\n",
       " <pykafka.partition.Partition at 0x115d82fd0 (id=5)>}"
      ]
     },
     "execution_count": 86,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "balanced_consumer3._partitions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "m= balanced_consumer.consume(block=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'foo: 5000006'"
      ]
     },
     "execution_count": 62,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "m.value"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}