Skip to content

Kafka improve logging #383

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
CarlosGamero merged 2 commits into from
Dec 16, 2025
Merged

Kafka improve logging #383

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ee959b5
Reverting prev changes
CarlosGamero Dec 16, 2025
5b8c49f
Improving commit logs
CarlosGamero Dec 16, 2025
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
83 changes: 0 additions & 83 deletions packages/kafka/lib/AbstractKafkaConsumer.spec.ts
View file
Open in desktop

This file was deleted.

20 changes: 11 additions & 9 deletions packages/kafka/lib/AbstractKafkaConsumer.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -228,12 +228,10 @@ export abstract class AbstractKafkaConsumer<
stream: KafkaMessageBatchStream<DeserializedMessage<SupportedMessageValues<TopicsConfig>>>,
): Promise<void> {
for await (const messageBatch of stream) {
this.consumerStream?.pause()
await this.consume(
messageBatch.topic,
messageBatch.messages as DeserializedMessage<SupportedMessageValues<TopicsConfig>>,
)
this.consumerStream?.resume()
}
}

Expand Down Expand Up @@ -434,9 +432,8 @@ export abstract class AbstractKafkaConsumer<

private commit(messageOrBatch: MessageOrBatch<SupportedMessageValues<TopicsConfig>>) {
if (Array.isArray(messageOrBatch)) {
if (messageOrBatch.length === 0) {
return Promise.resolve()
}
if (messageOrBatch.length === 0) return Promise.resolve()

// biome-ignore lint/style/noNonNullAssertion: we check the length above
return this.commitMessage(messageOrBatch[messageOrBatch.length - 1]!)
} else {
Expand All @@ -445,13 +442,18 @@ export abstract class AbstractKafkaConsumer<
}

private async commitMessage(message: DeserializedMessage<SupportedMessageValues<TopicsConfig>>) {
const logDetails = {
topic: message.topic,
offset: message.offset,
timestamp: message.timestamp,
}
this.logger.debug(logDetails, 'Trying to commit message')

try {
this.logger.debug(
{ topic: message.topic, offset: message.offset, timestamp: message.timestamp },
'Trying to commit message',
)
await message.commit()
this.logger.debug(logDetails, 'Message committed successfully')
} catch (error) {
this.logger.debug(logDetails, 'Message commit failed')
if (error instanceof ResponseError) return this.handleResponseErrorOnCommit(error)
throw error
}
Expand Down
Loading