Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Improve offset commit handling #775

Merged
merged 7 commits into from
Jun 23, 2020
Merged

Improve offset commit handling #775

merged 7 commits into from
Jun 23, 2020

Commits on Jun 23, 2020

  1. Only call committedOffsets() once 

    This function "does something", and is getting called inside a tight loop inside a critical
area. Essentially we can avoid the call overhead and the condition check in each invocation
here.
    @ankon
    ankon committed Jun 23, 2020
    Configuration menu
    Copy the full SHA
    fdcec54 View commit details
    Browse the repository at this point in the history

  2. Work on the complete partition offsets array at once 

    This skips a bunch of redundant property lookups: We can grab the needed
partition offsets array immediate for each topic, and then have the subtraction
logic simply process each partition of that.
    @ankon
    ankon committed Jun 23, 2020
    Configuration menu
    Copy the full SHA
    922880a View commit details
    Browse the repository at this point in the history

  3. Don't bother applying a regular expression to what should be a number 

    KafkaJS keeps track of offsets as strings and Longs to avoid problems with 64bit offsets. This
does require frequent string->Long and Long->string conversions so that the Long parts don't leak
into the interface.

In the case of offsets in the offset manager we might also have undefined/null or empty offset
strings, and isInvalidOffset needs to be able to detect these. But: We don't expect to have arbitrary
strings here, and figuring out whether a thing is undefined/null/'' doesn't need a regular expression
that checks the complete string -- a simple '!offset' is just fine.
    @ankon
    ankon committed Jun 23, 2020
    Configuration menu
    Copy the full SHA
    f73ac9a View commit details
    Browse the repository at this point in the history

  4. Check for "offset is negative" directly 

    Long.compare does a lot more than what we need here: We just want to know whether
the offset is negative (-1 for undefined, or potentially one of the special values).
    @ankon
    ankon committed Jun 23, 2020
    Configuration menu
    Copy the full SHA
    45af98d View commit details
    Browse the repository at this point in the history

  5. Use isInvalidOffset to check whether the committed offset is known 

    In some cases it seems possible that the committed offset isn't '-1', but rather undefined:
~~~~
kafkajs.Runner - Error when calling eachBatch ({"topic":"carrier-events","partition":0,"offset":"13435","stack":"TypeError: Cannot read property 'low' of undefined\n    at Function.fromValue (/app/node_modules/long/src/long.js:286:25)\n    at subtractOffsets (/app/node_modules/@collaborne/kafkajs/src/consumer/offsetManager/index.js:110:44)\n    at subtractPartitionOffsets (/app/node_modules/@collaborne/kafkajs/src/consumer/offsetManager/index.js:114:7)\n    at /app/node_modules/@collaborne/kafkajs/src/consumer/offsetManager/index.js:120:44\n    at Array.map (<anonymous>)\n    at subtractTopicOffsets (/app/node_modules/@collaborne/kafkajs/src/consumer/offsetManager/index.js:120:27)\n    at Array.map (<anonymous>)\n    at OffsetManager.countResolvedOffsets (/app/node_modules/@collaborne/kafkajs/src/consumer/offsetManager/index.js:122:37)\n    at OffsetManager.commitOffsetsIfNecessary (/app/node_modules/@collaborne/kafkajs/src/consumer/offsetManager/index.js:189:12)\n    at ConsumerGroup.commitOffsetsIfNecessary (/app/node_modules/@collaborne/kafkajs/src/consumer/consumerGroup.js:295:30)\n    at commitOffsetsIfNecessary (/app/node_modules/@collaborne/kafkajs/src/consumer/runner.js:235:34)\n    at messageGenerator (/app/node_modules/@collaborne/event-stream-library/src/client/shared-consumer.ts:341:10)\n    at runMicrotasks (<anonymous>)\n    at runNextTicks (internal/process/task_queues.js:62:5)\n    at processImmediate (internal/timers.js:429:9)\n    at process.topLevelDomainCallback (domain.js:137:15)"})
~~~~

From what I can see this might be related to `seek` and `resolve` operations overlapping, which
is possible: `seek` is only possible when the consumer is running, and when the consumer is running
it will do resolving/committing. This is likely a bigger issue when the consumer group contains
many topics.

After reading the code for quite a while I think the fix here is trivial: Instead of checking
explicitly for `-1` we simply should use the helper function created for that purpose.
    @ankon
    ankon committed Jun 23, 2020
    Configuration menu
    Copy the full SHA
    3faa4ff View commit details
    Browse the repository at this point in the history

  6. Use isInvalidOffset in nextOffset for consistency

    @ankon
    ankon committed Jun 23, 2020
    Configuration menu
    Copy the full SHA
    a9f8fe8 View commit details
    Browse the repository at this point in the history

  7. Fix isInvalidOffset for literal 0s 

    Pointed-out-by: @goriunov
    @ankon
    ankon committed Jun 23, 2020
    Configuration menu
    Copy the full SHA
    f478c69 View commit details
    Browse the repository at this point in the history