{"action":"created","issue":{"url":"https://api.github.com/repos/tulios/kafkajs/issues/122","repository_url":"https://api.github.com/repos/tulios/kafkajs","labels_url":"https://api.github.com/repos/tulios/kafkajs/issues/122/labels{/name}","comments_url":"https://api.github.com/repos/tulios/kafkajs/issues/122/comments","events_url":"https://api.github.com/repos/tulios/kafkajs/issues/122/events","html_url":"https://github.com/tulios/kafkajs/issues/122","id":354416527,"node_id":"MDU6SXNzdWUzNTQ0MTY1Mjc=","number":122,"title":"Add custom offset committing interface for consumer/runner","user":{"login":"bwdeng","id":2292986,"node_id":"MDQ6VXNlcjIyOTI5ODY=","avatar_url":"https://avatars3.githubusercontent.com/u/2292986?v=4","gravatar_id":"","url":"https://api.github.com/users/bwdeng","html_url":"https://github.com/bwdeng","followers_url":"https://api.github.com/users/bwdeng/followers","following_url":"https://api.github.com/users/bwdeng/following{/other_user}","gists_url":"https://api.github.com/users/bwdeng/gists{/gist_id}","starred_url":"https://api.github.com/users/bwdeng/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/bwdeng/subscriptions","organizations_url":"https://api.github.com/users/bwdeng/orgs","repos_url":"https://api.github.com/users/bwdeng/repos","events_url":"https://api.github.com/users/bwdeng/events{/privacy}","received_events_url":"https://api.github.com/users/bwdeng/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":3,"created_at":"2018-08-27T17:59:27Z","updated_at":"2018-08-28T20:02:21Z","closed_at":null,"author_association":"NONE","body":"**Feature**\r\n\r\nCurrently, the consumer only supports committing the resolved offset. It would be nice to support custom offset committing.\r\n\r\n**Use case**\r\n\r\nWe are trying to do aggregation based on kafkajs, but the size of the working set is not fixed. For example, an input stream would look like this:\r\n\r\n```\r\nX0 -- X1 -- X2 -- X3 -- Y0 -- Y1 -- Z0 -- Z1 -- Z2 -- ...\r\n```\r\n\r\nWe try to aggrege events of the same kind (agg([X0, X1, X2, X3]), agg([Y0, Y1]), agg([Z0, Z1, Z2, ...])), and only committing offset when the aggregated result has been saved. The problem is that until seeing a Y event, we don't know if all the X events have arrived. However, when we saw Y0, the resolved offset has already passed X3.\r\n\r\nHaving the ability to commit custom offsets would allow us to properly track the working progress. Thanks!\r\n"},"comment":{"url":"https://api.github.com/repos/tulios/kafkajs/issues/comments/416721310","html_url":"https://github.com/tulios/kafkajs/issues/122#issuecomment-416721310","issue_url":"https://api.github.com/repos/tulios/kafkajs/issues/122","id":416721310,"node_id":"MDEyOklzc3VlQ29tbWVudDQxNjcyMTMxMA==","user":{"login":"bwdeng","id":2292986,"node_id":"MDQ6VXNlcjIyOTI5ODY=","avatar_url":"https://avatars3.githubusercontent.com/u/2292986?v=4","gravatar_id":"","url":"https://api.github.com/users/bwdeng","html_url":"https://github.com/bwdeng","followers_url":"https://api.github.com/users/bwdeng/followers","following_url":"https://api.github.com/users/bwdeng/following{/other_user}","gists_url":"https://api.github.com/users/bwdeng/gists{/gist_id}","starred_url":"https://api.github.com/users/bwdeng/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/bwdeng/subscriptions","organizations_url":"https://api.github.com/users/bwdeng/orgs","repos_url":"https://api.github.com/users/bwdeng/repos","events_url":"https://api.github.com/users/bwdeng/events{/privacy}","received_events_url":"https://api.github.com/users/bwdeng/received_events","type":"User","site_admin":false},"created_at":"2018-08-28T20:02:21Z","updated_at":"2018-08-28T20:02:21Z","author_association":"NONE","body":"Exactly, I was hoping that there's a way to commit the offset of a message outside of its original batch callback.\r\n\r\nAt first, I tried updating the offset through the admin interface. But got `The coordinator is not aware of this member` error. It seems like only the active member can update the offsets of their assigned partitions.\r\n\r\nCurrently, I have a flaky patch which exposes the underlying ConsumerGroup used by the Runner, and use it to update offsets:\r\n\r\n```typescript\r\npublic async commit(topic: string, partition: number, offset: string): Promise<boolean> {\r\n    const { groupId, generationId, memberId, coordinator } = this.consumerGroup; // the underlying consumer group\r\n    if (coordinator && coordinator.isConnected()) {\r\n        await coordinator.offsetCommit({\r\n            groupId,\r\n            memberId,\r\n            groupGenerationId: generationId,\r\n            topics: [\r\n                {\r\n                    topic,\r\n                    partitions: [{ partition, offset }],\r\n                },\r\n            ],\r\n        });\r\n        return true;\r\n    }\r\n    return false;\r\n}\r\n```\r\n\r\nBut I don't think this is the proper way to do it. Also, I had to patch the `fetch` function [here](https://github.com/tulios/kafkajs/blob/e351b1618e1d1828000472fae86c2f4889a1a35c/src/consumer/runner.js#L244) as `commitOffsetsIfNecessary` to disable the auto-offset committing after each fetch.\r\n\r\nSo do you suggest that we should use something like:\r\n\r\n```javascript\r\nasync function commit(offset) {\r\n    const currentOffset = getResolvedOffset() // backup current offset\r\n    await resolveOffset(offset) // resolve offset for committing only\r\n    await commitOffsetsIfNecessary() // commit offset\r\n    await resolveOffset(currentOffset) // restore the original offset\r\n}\r\n```\r\n\r\nThanks!"}}
{"actor":{"display_login":"bwdeng"}}