Skip to content

MINOR: Reduce memory allocation in ClientTelemetryReporter.java #15402

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

Merged
divijvaidya merged 7 commits into from
Mar 8, 2024
Merged

MINOR: Reduce memory allocation in ClientTelemetryReporter.java #15402

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
a27f464
Update ClientTelemetryReporter.java
testn Feb 21, 2024
68562e4
Merge branch 'apache:trunk' into perf-reduce-mem
testn Feb 21, 2024
a1efb2b
Merge branch 'apache:trunk' into perf-reduce-mem
testn Feb 21, 2024
a7f786e
Update suppression rules
testn Feb 21, 2024
ee7e0e3
Merge branch 'apache:trunk' into perf-reduce-mem
testn Feb 22, 2024
76d58cc
Merge branch 'apache:trunk' into perf-reduce-mem
testn Mar 1, 2024
59f36f6
Merge branch 'apache:trunk' into perf-reduce-mem
testn Mar 4, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion checkstyle/suppressions.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,7 @@
files="(Utils|Topic|KafkaLZ4BlockOutputStream|AclData|JoinGroupRequest).java"/>

<suppress checks="CyclomaticComplexity"
files="(AbstractFetch|ConsumerCoordinator|CommitRequestManager|FetchCollector|OffsetFetcherUtils|KafkaProducer|Sender|ConfigDef|KerberosLogin|AbstractRequest|AbstractResponse|Selector|SslFactory|SslTransportLayer|SaslClientAuthenticator|SaslClientCallbackHandler|SaslServerAuthenticator|AbstractCoordinator|TransactionManager|AbstractStickyAssignor|DefaultSslEngineFactory|Authorizer|RecordAccumulator|MemoryRecords|FetchSessionHandler|MockAdminClient).java"/>
files="(AbstractFetch|ClientTelemetryReporter|ConsumerCoordinator|CommitRequestManager|FetchCollector|OffsetFetcherUtils|KafkaProducer|Sender|ConfigDef|KerberosLogin|AbstractRequest|AbstractResponse|Selector|SslFactory|SslTransportLayer|SaslClientAuthenticator|SaslClientCallbackHandler|SaslServerAuthenticator|AbstractCoordinator|TransactionManager|AbstractStickyAssignor|DefaultSslEngineFactory|Authorizer|RecordAccumulator|MemoryRecords|FetchSessionHandler|MockAdminClient).java"/>

@divijvaidya divijvaidya Feb 21, 2024

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.

why are we suppressing this? Ideally we would like to follow the same styling as rest of the code and hence fix the checkstyle failures.

@testn testn Feb 22, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

How about removing the trace out completely? I'm not a fan of calling trace in such a tight loop

@divijvaidya divijvaidya Feb 22, 2024

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.

I am not sure who might be relying on this log and I am not aware of the original motivation of adding this. I would prefer to keep the current state and only focus on perf improvement in this PR.

@testn testn Feb 22, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

If you prefer to keep the trace logic, the added code is purely to reduce memory allocation but it comes with extra baggage for additional cyclomatic complexity and hence the suppression

@divijvaidya divijvaidya Feb 22, 2024

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.

Understood. Thanks. This answers my question on my we are suppressing it.


<suppress checks="JavaNCSS"
files="(AbstractRequest|AbstractResponse|KerberosLogin|WorkerSinkTaskTest|TransactionManagerTest|SenderTest|KafkaAdminClient|ConsumerCoordinatorTest|KafkaAdminClientTest|KafkaRaftClientTest).java"/>
Expand Down
19 changes: 11 additions & 8 deletions ...ts/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -265,8 +265,8 @@ class DefaultClientTelemetrySender implements ClientTelemetrySender {
These are the lower and upper bounds of the jitter that we apply to the initial push
telemetry API call. This helps to avoid a flood of requests all coming at the same time.
*/
private final static double INITIAL_PUSH_JITTER_LOWER = 0.5;
private final static double INITIAL_PUSH_JITTER_UPPER = 1.5;
private static final double INITIAL_PUSH_JITTER_LOWER = 0.5;
private static final double INITIAL_PUSH_JITTER_UPPER = 1.5;

private final ReadWriteLock lock = new ReentrantReadWriteLock();
private final Condition subscriptionLoaded = lock.writeLock().newCondition();
Expand Down Expand Up @@ -325,6 +325,7 @@ public long timeToNextUpdate(long requestTimeoutMs) {
final long timeMs;
final String apiName;
final String msg;
final boolean isTraceEnabled = log.isTraceEnabled();

switch (localState) {
case SUBSCRIPTION_IN_PROGRESS:
Expand All @@ -336,33 +337,35 @@ public long timeToNextUpdate(long requestTimeoutMs) {
*/
apiName = (localState == ClientTelemetryState.SUBSCRIPTION_IN_PROGRESS) ? ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS.name : ApiKeys.PUSH_TELEMETRY.name;
timeMs = requestTimeoutMs;
msg = String.format("the remaining wait time for the %s network API request, as specified by %s", apiName, CommonClientConfigs.REQUEST_TIMEOUT_MS_CONFIG);
msg = isTraceEnabled ? "" : String.format("the remaining wait time for the %s network API request, as specified by %s", apiName, CommonClientConfigs.REQUEST_TIMEOUT_MS_CONFIG);
break;
case TERMINATING_PUSH_IN_PROGRESS:
timeMs = Long.MAX_VALUE;
msg = String.format("the terminating push is in progress, disabling telemetry for further requests");
msg = isTraceEnabled ? "" : "the terminating push is in progress, disabling telemetry for further requests";
break;
case TERMINATING_PUSH_NEEDED:
timeMs = 0;
msg = String.format("the client should try to submit the final %s network API request ASAP before closing", ApiKeys.PUSH_TELEMETRY.name);
msg = isTraceEnabled ? "" : String.format("the client should try to submit the final %s network API request ASAP before closing", ApiKeys.PUSH_TELEMETRY.name);
break;
case SUBSCRIPTION_NEEDED:
case PUSH_NEEDED:
apiName = (localState == ClientTelemetryState.SUBSCRIPTION_NEEDED) ? ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS.name : ApiKeys.PUSH_TELEMETRY.name;
long timeRemainingBeforeRequest = localLastRequestMs + localIntervalMs - nowMs;
if (timeRemainingBeforeRequest <= 0) {
timeMs = 0;
msg = String.format("the wait time before submitting the next %s network API request has elapsed", apiName);
msg = isTraceEnabled ? "" : String.format("the wait time before submitting the next %s network API request has elapsed", apiName);
} else {
timeMs = timeRemainingBeforeRequest;
msg = String.format("the client will wait before submitting the next %s network API request", apiName);
msg = isTraceEnabled ? "" : String.format("the client will wait before submitting the next %s network API request", apiName);
}
break;
default:
throw new IllegalStateException("Unknown telemetry state: " + localState);
}

log.trace("For telemetry state {}, returning the value {} ms; {}", localState, timeMs, msg);

@ijuma ijuma Mar 9, 2024

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.

Hmm, unlike what the PR says, there is no string generated if the log is disabled. The only allocation is the array for passing the parameters. No harm in this change, but I wanted to clarify.

@ijuma ijuma Mar 9, 2024

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.

More details here https://www.baeldung.com/java-string-pool

if (isTraceEnabled) {
log.trace("For telemetry state {}, returning the value {} ms; {}", localState, timeMs, msg);
}
return timeMs;
}

Expand Down