Skip to content

Kafka-14743: update request metrics after callback#13297

Merged
showuon merged 4 commits into
apache:trunkfrom
showuon:KAFKA-14743_2
Feb 26, 2023
Merged

Kafka-14743: update request metrics after callback#13297
showuon merged 4 commits into
apache:trunkfrom
showuon:KAFKA-14743_2

Conversation

@showuon

@showuon showuon commented Feb 24, 2023

Copy link
Copy Markdown
Member

Currently, the kafka.network:type=RequestMetrics,name=MessageConversionsTimeMs,request=Fetch will not get updated because the request metrics is recorded BEFORE the messageConversions metrics value updated. That means, even if we updated the messageConversions metrics value, the request metrics will never reflect the update. This patch fixes it by updating the request metric after callback completed, so that the messageConversions metric value can be updated correctly.

Committer Checklist (excluded from commit message)

  • Verify design and implementation
  • Verify test coverage and CI build status
  • Verify documentation (including upgrade notes)

@showuon

showuon commented Feb 24, 2023

Copy link
Copy Markdown
Member Author

@chia7712 , please take a look. Thanks.

@divijvaidya divijvaidya left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Minor nit comment. The change in itself looks good.

}

def testV1Fetch(isFollowerFetch: Boolean): Unit = {
val fetchMessageConversionsTimeMsMetricName = s"$MessageConversionsTimeMs,request=Fetch"

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit

Is it possible to generate this string with some constants? e.g. we can use RequestMetrics.MessageConversionsTimeMs

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, we already used the constants from RequestMetrics.MessageConversionsTimeMs. It's just the fetch/produce request is another tag name that needed to be filtered out. Thanks.

@chia7712 chia7712 left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@showuon LGTM. nice finding. a minor comment is left.


TestUtils.waitUntilTrue(() => TestUtils.metersCount(fetchMessageConversionsTimeMsMetricName) > initialFetchMessageConversionsTimeMs,
s"The `MessageConversionsTimeMs` in fetch request metric count is not incremented after 5 seconds. " +
s"init: $initialFetchMessageConversionsTimeMs final: ${TestUtils.metersCount(fetchMessageConversionsTimeMsMetricName)}", 5000)

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you add tea for TemporaryMemoryBytes? It seems to me TemporaryMemoryBytes also get fixed by this patch .

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point. Updated.

@chia7712 chia7712 left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM

@showuon

showuon commented Feb 26, 2023

Copy link
Copy Markdown
Member Author

Failed tests are unrelated:

    Build / JDK 8 and Scala 2.12 / kafka.admin.DescribeConsumerGroupTest.testDescribeExistingGroupWithNoMembers()
    Build / JDK 8 and Scala 2.12 / kafka.server.KafkaServerKRaftRegistrationTest.[1] Type=ZK, Name=testRegisterZkBrokerInKraft, MetadataVersion=3.4-IV0, Security=PLAINTEXT
    Build / JDK 17 and Scala 2.13 / org.apache.kafka.clients.consumer.internals.EagerConsumerCoordinatorTest.testOutdatedCoordinatorAssignment()
    Build / JDK 17 and Scala 2.13 / org.apache.kafka.streams.integration.EosIntegrationTest.shouldWriteLatestOffsetsToCheckpointOnShutdown[exactly_once]
    Build / JDK 17 and Scala 2.13 / org.apache.kafka.streams.processor.internals.DefaultStateUpdaterTest.shouldPauseStandbyTaskAndNotTransitToUpdateStandbyAgain()
    Build / JDK 11 and Scala 2.13 / org.apache.kafka.streams.processor.internals.DefaultStateUpdaterTest.shouldPauseStandbyTaskAndNotTransitToUpdateStandbyAgain()
    Build / JDK 11 and Scala 2.13 / org.apache.kafka.streams.processor.internals.DefaultStateUpdaterTest.shouldPauseStandbyTaskAndNotTransitToUpdateStandbyAgain()

@showuon
showuon merged commit ae67257 into apache:trunk Feb 26, 2023
showuon added a commit that referenced this pull request Feb 26, 2023
Currently, the kafka.network:type=RequestMetrics,name=MessageConversionsTimeMs,request=Fetch will not get updated because the request metrics is recorded BEFORE the messageConversions metrics value updated. That means, even if we updated the messageConversions metrics value, the request metrics will never reflect the update. This patch fixes it by updating the request metric after callback completed, so that the messageConversions metric value can be updated correctly.

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>, Divij Vaidya <diviv@amazon.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants