{"action":"created","comment":{"url":"https://api.github.com/repos/spring-projects/spring-kafka/pulls/comments/55277821","id":55277821,"diff_hunk":"@@ -199,256 +183,314 @@ else if (messageListener instanceof MessageListener) {\n \t\t\tthrow new IllegalStateException(\"messageListener must be 'MessageListener' \"\n \t\t\t\t\t+ \"or 'AcknowledgingMessageListener', not \" + messageListener.getClass().getName());\n \t\t}\n-\t\tConsumer<K, V> consumer = this.consumerFactory.createConsumer();\n-\t\tConsumerRebalanceListener rebalanceListener = new ConsumerRebalanceListener() {\n-\n-\t\t\t@Override\n-\t\t\tpublic void onPartitionsRevoked(Collection<TopicPartition> partitions) {\n-\t\t\t\tlogger.info(\"partitions revoked:\" + partitions);\n-\t\t\t}\n-\n-\t\t\t@Override\n-\t\t\tpublic void onPartitionsAssigned(Collection<TopicPartition> partitions) {\n-\t\t\t\tassignedPartitions = partitions;\n-\t\t\t\tlogger.info(\"partitions assigned:\" + partitions);\n-\t\t\t}\n-\n-\t\t};\n-\t\tif (this.partitions == null) {\n-\t\t\tif (topicPattern != null) {\n-\t\t\t\tconsumer.subscribe(topicPattern, rebalanceListener);\n-\t\t\t}\n-\t\t\telse {\n-\t\t\t\tconsumer.subscribe(Arrays.asList(topics), rebalanceListener);\n-\t\t\t}\n-\t\t}\n-\t\telse {\n-\t\t\tList<TopicPartition> topicPartitions = Arrays.asList(partitions);\n-\t\t\tthis.definedPartitions = topicPartitions;\n-\t\t\tconsumer.assign(topicPartitions);\n-\t\t}\n-\t\tthis.consumer = consumer;\n \t\tif (getTaskExecutor() == null) {\n \t\t\tsetTaskExecutor(\n \t\t\t\t\tnew SimpleAsyncTaskExecutor(getBeanName() == null ? \"kafka-\" : (getBeanName() + \"-kafka-\")));\n \t\t}\n-\t\tgetTaskExecutor().execute(this);\n+\t\tthis.listenerConsumer = new ListenerConsumer(this.listener, this.acknowledgingMessageListener,\n+\t\t\t\tthis.resetStrategy, this.recentOffset);\n+\t\tgetTaskExecutor().execute(this.listenerConsumer);\n \t}\n \n \t@Override\n \tprotected void doStop() {\n \t\tif (isRunning()) {\n \t\t\tsetRunning(false);\n-\t\t\tthis.consumer.wakeup();\n+\t\t\tthis.listenerConsumer.consumer.wakeup();\n \t\t}\n \t}\n \n-\t@Override\n-\tpublic void run() {\n-\t\tint count = 0;\n-\t\tlong last = System.currentTimeMillis();\n-\t\tlong now;\n-\t\tif (isRunning() && this.definedPartitions != null) {\n-\t\t\tinitPartitionsIfNeeded();\n-\t\t}\n-\t\tfinal AckMode ackMode = getAckMode();\n-\t\twhile (isRunning()) {\n-\t\t\ttry {\n-\t\t\t\tif (logger.isTraceEnabled()) {\n-\t\t\t\t\tlogger.trace(\"Polling...\");\n+\n+\n+\tprivate class ListenerConsumer implements SchedulingAwareRunnable {\n+\n+\t\tprivate final Log logger = LogFactory.getLog(this.getClass());\n+\n+\t\tprivate final CommitCallback callback = new CommitCallback();\n+\n+\t\tprivate final Consumer<K, V> consumer;\n+\n+\t\tprivate final ConcurrentMap<String, ConcurrentMap<Integer, Long>> manualOffsets = new ConcurrentHashMap<>();\n+\n+\t\tprivate final Map<String, Map<Integer, Long>> offsets = new HashMap<String, Map<Integer, Long>>();\n+\n+\t\tprivate final MessageListener<K, V> listener;\n+\n+\t\tprivate final AcknowledgingMessageListener<K, V> acknowledgingMessageListener;\n+\n+\t\tprivate final ContainerOffsetResetStrategy resetStrategy;\n+\n+\t\tprivate final long recentOffset;\n+\n+\t\tprivate final boolean autoCommit = consumerFactory.isAutoCommit();\n+\n+\t\tprivate Thread consumerThread;\n+\n+\t\tprivate volatile Collection<TopicPartition> definedPartitions;\n+\n+\t\tprivate volatile Collection<TopicPartition> assignedPartitions;\n+\n+\t\tpublic ListenerConsumer(MessageListener<K, V> listener, AcknowledgingMessageListener<K, V> ackListener,\n+\t\t\t\tContainerOffsetResetStrategy resetStrategy, long recentOffset) {\n+\t\t\tAssert.state(getAckMode().equals(AckMode.MANUAL) || getAckMode().equals(AckMode.MANUAL_IMMEDIATE)\n+\t\t\t\t\t? !this.autoCommit : true,\n+\t\t\t\t\"Consumer cannot be configured for auto commit for ackMode \" + getAckMode());\n+\t\t\tConsumer<K, V> consumer = consumerFactory.createConsumer();\n+\t\t\tConsumerRebalanceListener rebalanceListener = new ConsumerRebalanceListener() {\n+\n+\t\t\t\t@Override\n+\t\t\t\tpublic void onPartitionsRevoked(Collection<TopicPartition> partitions) {\n+\t\t\t\t\tlogger.info(\"partitions revoked:\" + partitions);\n \t\t\t\t}\n-\t\t\t\tConsumerRecords<K, V> records = consumer.poll(getPollTimeout());\n-\t\t\t\tif (records != null) {\n-\t\t\t\t\tcount += records.count();\n-\t\t\t\t\tif (logger.isDebugEnabled()) {\n-\t\t\t\t\t\tlogger.debug(\"Received: \" + records.count() + \" records\");\n+\n+\t\t\t\t@Override\n+\t\t\t\tpublic void onPartitionsAssigned(Collection<TopicPartition> partitions) {\n+\t\t\t\t\tassignedPartitions = partitions;\n+\t\t\t\t\tlogger.info(\"partitions assigned:\" + partitions);\n+\t\t\t\t}\n+\n+\t\t\t};\n+\t\t\tif (partitions == null) {\n+\t\t\t\tif (topicPattern != null) {\n+\t\t\t\t\tconsumer.subscribe(topicPattern, rebalanceListener);\n+\t\t\t\t}\n+\t\t\t\telse {\n+\t\t\t\t\tconsumer.subscribe(Arrays.asList(topics), rebalanceListener);\n+\t\t\t\t}\n+\t\t\t}\n+\t\t\telse {\n+\t\t\t\tList<TopicPartition> topicPartitions = Arrays.asList(partitions);\n+\t\t\t\tthis.definedPartitions = topicPartitions;\n+\t\t\t\tconsumer.assign(topicPartitions);\n+\t\t\t}\n+\t\t\tthis.consumer = consumer;\n+\t\t\tthis.listener = listener;\n+\t\t\tthis.acknowledgingMessageListener = ackListener;\n+\t\t\tthis.resetStrategy = resetStrategy;\n+\t\t\tthis.recentOffset = recentOffset;\n+\t\t}\n+\n+\t\t@Override\n+\t\tpublic boolean isLongLived() {\n+\t\t\treturn true;\n+\t\t}\n+\n+\t\t@Override\n+\t\tpublic void run() {\n+\t\t\tthis.consumerThread = Thread.currentThread();\n+\t\t\tint count = 0;\n+\t\t\tlong last = System.currentTimeMillis();\n+\t\t\tlong now;\n+\t\t\tif (isRunning() && definedPartitions != null) {\n+\t\t\t\tinitPartitionsIfNeeded();\n+\t\t\t}\n+\t\t\tfinal AckMode ackMode = getAckMode();\n+\t\t\twhile (isRunning()) {\n+\t\t\t\ttry {\n+\t\t\t\t\tif (logger.isTraceEnabled()) {\n+\t\t\t\t\t\tlogger.trace(\"Polling...\");\n \t\t\t\t\t}\n-\t\t\t\t\tIterator<ConsumerRecord<K, V>> iterator = records.iterator();\n-\t\t\t\t\twhile (iterator.hasNext()) {\n-\t\t\t\t\t\tfinal ConsumerRecord<K, V> record = iterator.next();\n-\t\t\t\t\t\tinvokeListener(record);\n-\t\t\t\t\t\tif (!this.autoCommit && ackMode.equals(AckMode.RECORD)) {\n-\t\t\t\t\t\t\tthis.consumer.commitAsync(\n-\t\t\t\t\t\t\t\t\tCollections.singletonMap(new TopicPartition(record.topic(), record.partition()),\n-\t\t\t\t\t\t\t\t\t\t\tnew OffsetAndMetadata(record.offset())), callback);\n+\t\t\t\t\tConsumerRecords<K, V> records = consumer.poll(getPollTimeout());\n+\t\t\t\t\tif (records != null) {\n+\t\t\t\t\t\tcount += records.count();\n+\t\t\t\t\t\tif (logger.isDebugEnabled()) {\n+\t\t\t\t\t\t\tlogger.debug(\"Received: \" + records.count() + \" records\");\n \t\t\t\t\t\t}\n-\t\t\t\t\t}\n-\t\t\t\t\tif (!this.autoCommit) {\n-\t\t\t\t\t\tif (ackMode.equals(AckMode.BATCH)) {\n-\t\t\t\t\t\t\tif (!records.isEmpty()) {\n-\t\t\t\t\t\t\t\tthis.consumer.commitAsync(callback);\n+\t\t\t\t\t\tIterator<ConsumerRecord<K, V>> iterator = records.iterator();\n+\t\t\t\t\t\twhile (iterator.hasNext()) {\n+\t\t\t\t\t\t\tfinal ConsumerRecord<K, V> record = iterator.next();\n+\t\t\t\t\t\t\tinvokeListener(record);\n+\t\t\t\t\t\t\tif (!autoCommit && ackMode.equals(AckMode.RECORD)) {\n+\t\t\t\t\t\t\t\tthis.consumer.commitAsync(\n+\t\t\t\t\t\t\t\t\t\tCollections.singletonMap(new TopicPartition(record.topic(), record.partition()),\n+\t\t\t\t\t\t\t\t\t\t\t\tnew OffsetAndMetadata(record.offset() + 1)), this.callback);\n \t\t\t\t\t\t\t}\n \t\t\t\t\t\t}\n-\t\t\t\t\t\telse {\n-\t\t\t\t\t\t\tif (!ackMode.equals(AckMode.MANUAL)) {\n-\t\t\t\t\t\t\t\tupdatePendingOffsets(records);\n-\t\t\t\t\t\t\t}\n-\t\t\t\t\t\t\tboolean countExceeded = count >= getAckCount();\n-\t\t\t\t\t\t\tif (ackMode.equals(AckMode.COUNT) && countExceeded) {\n-\t\t\t\t\t\t\t\tcommitIfNecessary();\n-\t\t\t\t\t\t\t\tcount = 0;\n+\t\t\t\t\t\tif (!this.autoCommit) {\n+\t\t\t\t\t\t\tif (ackMode.equals(AckMode.BATCH)) {\n+\t\t\t\t\t\t\t\tif (!records.isEmpty()) {\n+\t\t\t\t\t\t\t\t\tthis.consumer.commitAsync(this.callback);\n+\t\t\t\t\t\t\t\t}\n \t\t\t\t\t\t\t}\n-\t\t\t\t\t\t\telse {\n-\t\t\t\t\t\t\t\tnow = System.currentTimeMillis();\n-\t\t\t\t\t\t\t\tboolean elapsed = now - last > getAckTime();\n-\t\t\t\t\t\t\t\tif (ackMode.equals(AckMode.TIME) && elapsed) {\n-\t\t\t\t\t\t\t\t\tcommitIfNecessary();\n-\t\t\t\t\t\t\t\t\tlast = now;\n+\t\t\t\t\t\t\telse if (!ackMode.equals(AckMode.MANUAL_IMMEDIATE)) {\n+\t\t\t\t\t\t\t\tif (!ackMode.equals(AckMode.MANUAL)) {\n+\t\t\t\t\t\t\t\t\tupdatePendingOffsets(records);\n \t\t\t\t\t\t\t\t}\n-\t\t\t\t\t\t\t\telse if ((ackMode.equals(AckMode.COUNT_TIME) || ackMode.equals(AckMode.MANUAL))\n-\t\t\t\t\t\t\t\t\t\t&& (elapsed || countExceeded)) {\n+\t\t\t\t\t\t\t\tboolean countExceeded = count >= getAckCount();\n+\t\t\t\t\t\t\t\tif (ackMode.equals(AckMode.COUNT) && countExceeded) {\n \t\t\t\t\t\t\t\t\tcommitIfNecessary();\n-\t\t\t\t\t\t\t\t\tlast = now;\n \t\t\t\t\t\t\t\t\tcount = 0;\n \t\t\t\t\t\t\t\t}\n+\t\t\t\t\t\t\t\telse {\n+\t\t\t\t\t\t\t\t\tnow = System.currentTimeMillis();\n+\t\t\t\t\t\t\t\t\tboolean elapsed = now - last > getAckTime();\n+\t\t\t\t\t\t\t\t\tif (ackMode.equals(AckMode.TIME) && elapsed) {\n+\t\t\t\t\t\t\t\t\t\tcommitIfNecessary();\n+\t\t\t\t\t\t\t\t\t\tlast = now;\n+\t\t\t\t\t\t\t\t\t}\n+\t\t\t\t\t\t\t\t\telse if ((ackMode.equals(AckMode.COUNT_TIME) || ackMode.equals(AckMode.MANUAL))\n+\t\t\t\t\t\t\t\t\t\t\t&& (elapsed || countExceeded)) {\n+\t\t\t\t\t\t\t\t\t\tcommitIfNecessary();\n+\t\t\t\t\t\t\t\t\t\tlast = now;\n+\t\t\t\t\t\t\t\t\t\tcount = 0;\n+\t\t\t\t\t\t\t\t\t}\n+\t\t\t\t\t\t\t\t}\n \t\t\t\t\t\t\t}\n \t\t\t\t\t\t}\n \t\t\t\t\t}\n+\t\t\t\t\telse {\n+\t\t\t\t\t\tif (logger.isDebugEnabled()) {\n+\t\t\t\t\t\t\tlogger.debug(\"No records\");\n+\t\t\t\t\t\t}\n+\t\t\t\t\t}\n \t\t\t\t}\n-\t\t\t\telse {\n-\t\t\t\t\tif (logger.isDebugEnabled()) {\n-\t\t\t\t\t\tlogger.debug(\"No records\");\n+\t\t\t\tcatch (WakeupException e) {\n+\t\t\t\t\t;\n+\t\t\t\t}\n+\t\t\t\tcatch (Exception e) {\n+\t\t\t\t\tif (getErrorHandler() != null) {\n+\t\t\t\t\t\tgetErrorHandler().handle(e, null);\n+\t\t\t\t\t}\n+\t\t\t\t\telse {\n+\t\t\t\t\t\tlogger.error(\"Container exception\", e);\n \t\t\t\t\t}\n \t\t\t\t}\n \t\t\t}\n+\t\t\tif (offsets.size() > 0) {\n+\t\t\t\tcommitIfNecessary();\n+\t\t\t}\n+\t\t\ttry {\n+\t\t\t\tthis.consumer.unsubscribe();\n+\t\t\t}\n \t\t\tcatch (WakeupException e) {\n \t\t\t\t;\n \t\t\t}\n-\t\t\tcatch (Exception e) {\n-\t\t\t\tif (getErrorHandler() != null) {\n-\t\t\t\t\tgetErrorHandler().handle(e, null);\n-\t\t\t\t}\n-\t\t\t\telse {\n-\t\t\t\t\tlogger.error(\"Container exception\", e);\n-\t\t\t\t}\n+\t\t\tthis.consumer.close();\n+\t\t\tif (logger.isInfoEnabled()) {\n+\t\t\t\tlogger.info(\"Consumer stopped\");\n \t\t\t}\n \t\t}\n-\t\tif (this.offsets.size() > 0) {\n-\t\t\tcommitIfNecessary();\n-\t\t}\n-\t\ttry {\n-\t\t\tthis.consumer.unsubscribe();\n-\t\t}\n-\t\tcatch (WakeupException e) {\n-\t\t\t;\n-\t\t}\n-\t\tthis.consumer.close();\n-\t\tif (logger.isInfoEnabled()) {\n-\t\t\tlogger.info(\"Consumer stopped\");\n-\t\t}\n-\t}\n \n-\tprivate void invokeListener(final ConsumerRecord<K, V> record) {\n-\t\ttry {\n-\t\t\tif (this.acknowledgingMessageListener != null) {\n-\t\t\t\tthis.acknowledgingMessageListener.onMessage(record, new Acknowledgment() {\n+\t\tprivate void invokeListener(final ConsumerRecord<K, V> record) {\n+\t\t\ttry {\n+\t\t\t\tif (this.acknowledgingMessageListener != null) {\n+\t\t\t\t\tthis.acknowledgingMessageListener.onMessage(record, new Acknowledgment() {\n \n-\t\t\t\t\t@Override\n-\t\t\t\t\tpublic void acknowledge() {\n-\t\t\t\t\t\tif (getAckMode().equals(AckMode.MANUAL)) {\n-\t\t\t\t\t\t\tupdateManualOffset(record);\n-\t\t\t\t\t\t}\n-\t\t\t\t\t\telse if (getAckMode().equals(AckMode.MANUAL_IMMEDIATE)) {\n-\t\t\t\t\t\t\tupdateManualOffset(record);\n-\t\t\t\t\t\t\tconsumer.wakeup();\n-\t\t\t\t\t\t}\n-\t\t\t\t\t\telse {\n-\t\t\t\t\t\t\tthrow new IllegalStateException(\"AckMode must be MANUAL or MANUAL_IMMEDIATE \"\n-\t\t\t\t\t\t\t\t\t+ \"for manual acks\");\n+\t\t\t\t\t\t@Override\n+\t\t\t\t\t\tpublic void acknowledge() {\n+\t\t\t\t\t\t\tif (getAckMode().equals(AckMode.MANUAL)) {\n+\t\t\t\t\t\t\t\tupdateManualOffset(record);\n+\t\t\t\t\t\t\t}\n+\t\t\t\t\t\t\telse if (getAckMode().equals(AckMode.MANUAL_IMMEDIATE)) {\n+\t\t\t\t\t\t\t\tif (Thread.currentThread().equals(consumerThread)) {\n+\t\t\t\t\t\t\t\t\tMap<TopicPartition, OffsetAndMetadata> commits = Collections.singletonMap(\n+\t\t\t\t\t\t\t\t\t\t\tnew TopicPartition(record.topic(), record.partition()),\n+\t\t\t\t\t\t\t\t\t\t\tnew OffsetAndMetadata(record.offset() + 1));\n+\t\t\t\t\t\t\t\t\tif (logger.isDebugEnabled()) {\n+\t\t\t\t\t\t\t\t\t\tlogger.debug(\"Committing: \" + commits);\n+\t\t\t\t\t\t\t\t\t}\n+\t\t\t\t\t\t\t\t\tconsumer.commitAsync(commits, callback);\n+\t\t\t\t\t\t\t\t}\n+\t\t\t\t\t\t\t\telse {\n+\t\t\t\t\t\t\t\t\tthrow new IllegalStateException(\n+\t\t\t\t\t\t\t\t\t\t\t\"With MANUAL_IMMEDIATE ack mode, acknowledget must be invoked on the \"\n+\t\t\t\t\t\t\t\t\t\t\t+ \"consumer thread\");","path":"spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java","position":405,"original_position":405,"commit_id":"ef06a4901e6bdf03cddc1c4307da58fd23c893e3","original_commit_id":"ef06a4901e6bdf03cddc1c4307da58fd23c893e3","user":{"login":"garyrussell","id":483832,"avatar_url":"https://avatars.githubusercontent.com/u/483832?v=3","gravatar_id":"","url":"https://api.github.com/users/garyrussell","html_url":"https://github.com/garyrussell","followers_url":"https://api.github.com/users/garyrussell/followers","following_url":"https://api.github.com/users/garyrussell/following{/other_user}","gists_url":"https://api.github.com/users/garyrussell/gists{/gist_id}","starred_url":"https://api.github.com/users/garyrussell/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/garyrussell/subscriptions","organizations_url":"https://api.github.com/users/garyrussell/orgs","repos_url":"https://api.github.com/users/garyrussell/repos","events_url":"https://api.github.com/users/garyrussell/events{/privacy}","received_events_url":"https://api.github.com/users/garyrussell/received_events","type":"User","site_admin":false},"body":"Right, that's why my previous (flawed) solution used wake.","created_at":"2016-03-07T21:45:19Z","updated_at":"2016-03-07T21:45:19Z","html_url":"https://github.com/spring-projects/spring-kafka/pull/14#discussion_r55277821","pull_request_url":"https://api.github.com/repos/spring-projects/spring-kafka/pulls/14","_links":{"self":{"href":"https://api.github.com/repos/spring-projects/spring-kafka/pulls/comments/55277821"},"html":{"href":"https://github.com/spring-projects/spring-kafka/pull/14#discussion_r55277821"},"pull_request":{"href":"https://api.github.com/repos/spring-projects/spring-kafka/pulls/14"}}},"pull_request":{"url":"https://api.github.com/repos/spring-projects/spring-kafka/pulls/14","id":61961130,"html_url":"https://github.com/spring-projects/spring-kafka/pull/14","diff_url":"https://github.com/spring-projects/spring-kafka/pull/14.diff","patch_url":"https://github.com/spring-projects/spring-kafka/pull/14.patch","issue_url":"https://api.github.com/repos/spring-projects/spring-kafka/issues/14","number":14,"state":"open","locked":false,"title":"Fix MANUAL_IMMEDIATE","user":{"login":"garyrussell","id":483832,"avatar_url":"https://avatars.githubusercontent.com/u/483832?v=3","gravatar_id":"","url":"https://api.github.com/users/garyrussell","html_url":"https://github.com/garyrussell","followers_url":"https://api.github.com/users/garyrussell/followers","following_url":"https://api.github.com/users/garyrussell/following{/other_user}","gists_url":"https://api.github.com/users/garyrussell/gists{/gist_id}","starred_url":"https://api.github.com/users/garyrussell/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/garyrussell/subscriptions","organizations_url":"https://api.github.com/users/garyrussell/orgs","repos_url":"https://api.github.com/users/garyrussell/repos","events_url":"https://api.github.com/users/garyrussell/events{/privacy}","received_events_url":"https://api.github.com/users/garyrussell/received_events","type":"User","site_admin":false},"body":"Resolves #12\r\n\r\nExecute the commit immediately, after asserting the callback is invoked on the consumer\r\nthread.\r\n\r\nRestructure consumer as an inner class.\r\n\r\n__Given the restructuring, it might be easier to review this with `?w=1`__","created_at":"2016-03-07T18:46:11Z","updated_at":"2016-03-07T21:45:19Z","closed_at":null,"merged_at":null,"merge_commit_sha":"8d5761def63cc4f158f350ec652f70dfb9a1015e","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/spring-projects/spring-kafka/pulls/14/commits","review_comments_url":"https://api.github.com/repos/spring-projects/spring-kafka/pulls/14/comments","review_comment_url":"https://api.github.com/repos/spring-projects/spring-kafka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/spring-projects/spring-kafka/issues/14/comments","statuses_url":"https://api.github.com/repos/spring-projects/spring-kafka/statuses/ef06a4901e6bdf03cddc1c4307da58fd23c893e3","head":{"label":"garyrussell:gh-12","ref":"gh-12","sha":"ef06a4901e6bdf03cddc1c4307da58fd23c893e3","user":{"login":"garyrussell","id":483832,"avatar_url":"https://avatars.githubusercontent.com/u/483832?v=3","gravatar_id":"","url":"https://api.github.com/users/garyrussell","html_url":"https://github.com/garyrussell","followers_url":"https://api.github.com/users/garyrussell/followers","following_url":"https://api.github.com/users/garyrussell/following{/other_user}","gists_url":"https://api.github.com/users/garyrussell/gists{/gist_id}","starred_url":"https://api.github.com/users/garyrussell/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/garyrussell/subscriptions","organizations_url":"https://api.github.com/users/garyrussell/orgs","repos_url":"https://api.github.com/users/garyrussell/repos","events_url":"https://api.github.com/users/garyrussell/events{/privacy}","received_events_url":"https://api.github.com/users/garyrussell/received_events","type":"User","site_admin":false},"repo":{"id":53343563,"name":"spring-kafka","full_name":"garyrussell/spring-kafka","owner":{"login":"garyrussell","id":483832,"avatar_url":"https://avatars.githubusercontent.com/u/483832?v=3","gravatar_id":"","url":"https://api.github.com/users/garyrussell","html_url":"https://github.com/garyrussell","followers_url":"https://api.github.com/users/garyrussell/followers","following_url":"https://api.github.com/users/garyrussell/following{/other_user}","gists_url":"https://api.github.com/users/garyrussell/gists{/gist_id}","starred_url":"https://api.github.com/users/garyrussell/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/garyrussell/subscriptions","organizations_url":"https://api.github.com/users/garyrussell/orgs","repos_url":"https://api.github.com/users/garyrussell/repos","events_url":"https://api.github.com/users/garyrussell/events{/privacy}","received_events_url":"https://api.github.com/users/garyrussell/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/garyrussell/spring-kafka","description":"","fork":true,"url":"https://api.github.com/repos/garyrussell/spring-kafka","forks_url":"https://api.github.com/repos/garyrussell/spring-kafka/forks","keys_url":"https://api.github.com/repos/garyrussell/spring-kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/garyrussell/spring-kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/garyrussell/spring-kafka/teams","hooks_url":"https://api.github.com/repos/garyrussell/spring-kafka/hooks","issue_events_url":"https://api.github.com/repos/garyrussell/spring-kafka/issues/events{/number}","events_url":"https://api.github.com/repos/garyrussell/spring-kafka/events","assignees_url":"https://api.github.com/repos/garyrussell/spring-kafka/assignees{/user}","branches_url":"https://api.github.com/repos/garyrussell/spring-kafka/branches{/branch}","tags_url":"https://api.github.com/repos/garyrussell/spring-kafka/tags","blobs_url":"https://api.github.com/repos/garyrussell/spring-kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/garyrussell/spring-kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/garyrussell/spring-kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/garyrussell/spring-kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/garyrussell/spring-kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/garyrussell/spring-kafka/languages","stargazers_url":"https://api.github.com/repos/garyrussell/spring-kafka/stargazers","contributors_url":"https://api.github.com/repos/garyrussell/spring-kafka/contributors","subscribers_url":"https://api.github.com/repos/garyrussell/spring-kafka/subscribers","subscription_url":"https://api.github.com/repos/garyrussell/spring-kafka/subscription","commits_url":"https://api.github.com/repos/garyrussell/spring-kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/garyrussell/spring-kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/garyrussell/spring-kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/garyrussell/spring-kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/garyrussell/spring-kafka/contents/{+path}","compare_url":"https://api.github.com/repos/garyrussell/spring-kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/garyrussell/spring-kafka/merges","archive_url":"https://api.github.com/repos/garyrussell/spring-kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/garyrussell/spring-kafka/downloads","issues_url":"https://api.github.com/repos/garyrussell/spring-kafka/issues{/number}","pulls_url":"https://api.github.com/repos/garyrussell/spring-kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/garyrussell/spring-kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/garyrussell/spring-kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/garyrussell/spring-kafka/labels{/name}","releases_url":"https://api.github.com/repos/garyrussell/spring-kafka/releases{/id}","deployments_url":"https://api.github.com/repos/garyrussell/spring-kafka/deployments","created_at":"2016-03-07T17:10:01Z","updated_at":"2016-03-07T17:10:02Z","pushed_at":"2016-03-07T21:31:58Z","git_url":"git://github.com/garyrussell/spring-kafka.git","ssh_url":"git@github.com:garyrussell/spring-kafka.git","clone_url":"https://github.com/garyrussell/spring-kafka.git","svn_url":"https://github.com/garyrussell/spring-kafka","homepage":null,"size":164,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"spring-projects:master","ref":"master","sha":"c0fe2c2fb8a5e05f3edf9a8b6a38dfa6a92a0f71","user":{"login":"spring-projects","id":317776,"avatar_url":"https://avatars.githubusercontent.com/u/317776?v=3","gravatar_id":"","url":"https://api.github.com/users/spring-projects","html_url":"https://github.com/spring-projects","followers_url":"https://api.github.com/users/spring-projects/followers","following_url":"https://api.github.com/users/spring-projects/following{/other_user}","gists_url":"https://api.github.com/users/spring-projects/gists{/gist_id}","starred_url":"https://api.github.com/users/spring-projects/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/spring-projects/subscriptions","organizations_url":"https://api.github.com/users/spring-projects/orgs","repos_url":"https://api.github.com/users/spring-projects/repos","events_url":"https://api.github.com/users/spring-projects/events{/privacy}","received_events_url":"https://api.github.com/users/spring-projects/received_events","type":"Organization","site_admin":false},"repo":{"id":52980493,"name":"spring-kafka","full_name":"spring-projects/spring-kafka","owner":{"login":"spring-projects","id":317776,"avatar_url":"https://avatars.githubusercontent.com/u/317776?v=3","gravatar_id":"","url":"https://api.github.com/users/spring-projects","html_url":"https://github.com/spring-projects","followers_url":"https://api.github.com/users/spring-projects/followers","following_url":"https://api.github.com/users/spring-projects/following{/other_user}","gists_url":"https://api.github.com/users/spring-projects/gists{/gist_id}","starred_url":"https://api.github.com/users/spring-projects/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/spring-projects/subscriptions","organizations_url":"https://api.github.com/users/spring-projects/orgs","repos_url":"https://api.github.com/users/spring-projects/repos","events_url":"https://api.github.com/users/spring-projects/events{/privacy}","received_events_url":"https://api.github.com/users/spring-projects/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/spring-projects/spring-kafka","description":"","fork":false,"url":"https://api.github.com/repos/spring-projects/spring-kafka","forks_url":"https://api.github.com/repos/spring-projects/spring-kafka/forks","keys_url":"https://api.github.com/repos/spring-projects/spring-kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/spring-projects/spring-kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/spring-projects/spring-kafka/teams","hooks_url":"https://api.github.com/repos/spring-projects/spring-kafka/hooks","issue_events_url":"https://api.github.com/repos/spring-projects/spring-kafka/issues/events{/number}","events_url":"https://api.github.com/repos/spring-projects/spring-kafka/events","assignees_url":"https://api.github.com/repos/spring-projects/spring-kafka/assignees{/user}","branches_url":"https://api.github.com/repos/spring-projects/spring-kafka/branches{/branch}","tags_url":"https://api.github.com/repos/spring-projects/spring-kafka/tags","blobs_url":"https://api.github.com/repos/spring-projects/spring-kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/spring-projects/spring-kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/spring-projects/spring-kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/spring-projects/spring-kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/spring-projects/spring-kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/spring-projects/spring-kafka/languages","stargazers_url":"https://api.github.com/repos/spring-projects/spring-kafka/stargazers","contributors_url":"https://api.github.com/repos/spring-projects/spring-kafka/contributors","subscribers_url":"https://api.github.com/repos/spring-projects/spring-kafka/subscribers","subscription_url":"https://api.github.com/repos/spring-projects/spring-kafka/subscription","commits_url":"https://api.github.com/repos/spring-projects/spring-kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/spring-projects/spring-kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/spring-projects/spring-kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/spring-projects/spring-kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/spring-projects/spring-kafka/contents/{+path}","compare_url":"https://api.github.com/repos/spring-projects/spring-kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/spring-projects/spring-kafka/merges","archive_url":"https://api.github.com/repos/spring-projects/spring-kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/spring-projects/spring-kafka/downloads","issues_url":"https://api.github.com/repos/spring-projects/spring-kafka/issues{/number}","pulls_url":"https://api.github.com/repos/spring-projects/spring-kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/spring-projects/spring-kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/spring-projects/spring-kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/spring-projects/spring-kafka/labels{/name}","releases_url":"https://api.github.com/repos/spring-projects/spring-kafka/releases{/id}","deployments_url":"https://api.github.com/repos/spring-projects/spring-kafka/deployments","created_at":"2016-03-02T17:01:33Z","updated_at":"2016-03-07T03:04:23Z","pushed_at":"2016-03-07T21:31:58Z","git_url":"git://github.com/spring-projects/spring-kafka.git","ssh_url":"git@github.com:spring-projects/spring-kafka.git","clone_url":"https://github.com/spring-projects/spring-kafka.git","svn_url":"https://github.com/spring-projects/spring-kafka","homepage":null,"size":143,"stargazers_count":3,"watchers_count":3,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":2,"mirror_url":null,"open_issues_count":16,"forks":2,"open_issues":16,"watchers":3,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/spring-projects/spring-kafka/pulls/14"},"html":{"href":"https://github.com/spring-projects/spring-kafka/pull/14"},"issue":{"href":"https://api.github.com/repos/spring-projects/spring-kafka/issues/14"},"comments":{"href":"https://api.github.com/repos/spring-projects/spring-kafka/issues/14/comments"},"review_comments":{"href":"https://api.github.com/repos/spring-projects/spring-kafka/pulls/14/comments"},"review_comment":{"href":"https://api.github.com/repos/spring-projects/spring-kafka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/spring-projects/spring-kafka/pulls/14/commits"},"statuses":{"href":"https://api.github.com/repos/spring-projects/spring-kafka/statuses/ef06a4901e6bdf03cddc1c4307da58fd23c893e3"}}}}