Add log channel to admin client #1448

k-raina · 2025-06-27T10:34:13Z

Summary

This PR adds support for forwarding librdkafka log events to a Go channel in the AdminClient, similar to the existing functionality in the Producer and Consumer clients. This allows users to programmatically consume and process log events generated by the underlying librdkafka instance used by the AdminClient.

Motivation

Addresses issue a lot of bogus LogEvents when using go-channel #1119 and admin client go logs #1127
where users requested the ability to receive log events from the AdminClient via a Go channel.
Enables better observability and integration with Go-based logging and monitoring systems.

Changes

AdminClient struct: Added an adminTermChan field to manage the lifecycle of the log polling goroutine.
NewAdminClient:
- Accepts go.logs.channel.enable and go.logs.channel in the ConfigMap.
- If enabled, sets up a log channel and starts a goroutine to forward log events.
Close method: Ensures the log polling goroutine is properly terminated when the AdminClient is closed.

Test

Unit Test

Added a test to verify that log events are received as expected.

Manual test

Test Code:

func main() {
	_, version := kafka.LibraryVersion()
	fmt.Printf("librdkafka version: %s\n", version)

	logsChan := make(chan kafka.LogEvent, 100)

	admin, err := kafka.NewAdminClient(&kafka.ConfigMap{
		"bootstrap.servers":      "localhost:9092", // Change as needed
		"debug":                  "all",
		"go.logs.channel.enable": true,
		"go.logs.channel":        logsChan,
	})
	if err != nil {
		fmt.Printf("Failed to create AdminClient: %v\n", err)
		return
	}
	defer admin.Close()

	fmt.Println("AdminClient created. Listening for log events... (Ctrl+C to exit)")

	sigchan := make(chan os.Signal, 1)
	signal.Notify(sigchan, os.Interrupt)

	go func() {
		for log := range logsChan {
			fmt.Printf("%s INFO kafka admin - LogEvent {\n", log.Timestamp.Format("15:04:05.00000"))
			fmt.Printf("  Name: '%s',\n", log.Name)
			fmt.Printf("  Tag: '%s',\n", log.Tag)
			fmt.Printf("  Message: '%s',\n", log.Message)
			fmt.Printf("  Level: %d,\n", log.Level)
			fmt.Printf("  Timestamp: Time {\n")
			fmt.Printf("   wall: %d,\n", log.Timestamp.UnixNano())
			fmt.Printf("   ext: %d,\n", log.Timestamp.Unix())
			fmt.Printf("   loc: %s\n", log.Timestamp.Location())
			fmt.Printf("  }\n")
			fmt.Printf(" } \n\n")
		}
	}()

	// Optionally, perform an admin operation to generate some logs
	_, _ = admin.GetMetadata(nil, true, 1000)

	<-sigchan
	fmt.Println("Exiting...")
}

Output

15:57:12.20329 INFO kafka admin - LogEvent {
  Name: 'rdkafka#producer-1',
  Tag: 'SEND',
  Message: '[thrd:localhost:9092/1]: localhost:9092/1: Sent MetadataRequest (v12, 26 bytes @ 0, CorrId 2)',
  Level: 7,
  Timestamp: Time {
   wall: 1751020032203293000,
   ext: 1751020032,
   loc: Local
  }
 } 

15:57:12.20335 INFO kafka admin - LogEvent {
  Name: 'rdkafka#producer-1',
  Tag: 'METADATA',
  Message: '[thrd:app]: localhost:9092/1: Request metadata for all topics: application requested',
  Level: 7,
  Timestamp: Time {
   wall: 1751020032203356000,
   ext: 1751020032,
   loc: Local
  }
 } 

15:57:12.20341 INFO kafka admin - LogEvent {
  Name: 'rdkafka#producer-1',
  Tag: 'SETBROKER',
  Message: '[thrd:main]: Setting telemetry broker to localhost:9092/1
',
  Level: 7,
  Timestamp: Time {
   wall: 1751020032203415000,
   ext: 1751020032,
   loc: Local
  }
 } 

15:57:12.20347 INFO kafka admin - LogEvent {
  Name: 'rdkafka#producer-1',
  Tag: 'SEND',
  Message: '[thrd:localhost:9092/1]: localhost:9092/1: Sent MetadataRequest (v12, 29 bytes @ 0, CorrId 3)',
  Level: 7,
  Timestamp: Time {
   wall: 1751020032203475000,
   ext: 1751020032,
   loc: Local
  }
 } 

15:57:12.20351 INFO kafka admin - LogEvent {
  Name: 'rdkafka#producer-1',
  Tag: 'GETSUBSCRIPTIONS',
  Message: '[thrd:main]: Sending GetTelemetryRequest',
  Level: 7,
  Timestamp: Time {
   wall: 1751020032203517000,
   ext: 1751020032,
   loc: Local
  }
 }

Copilot

Pull Request Overview

This PR adds support for forwarding librdkafka log events to a Go channel in the AdminClient, enabling programmatic log consumption for improved observability.

Adds a log channel setup in NewAdminClient and cleans it up in Close.
Introduces a new unit test (TestAdminClientLog) verifying the log event forwarding functionality.

Reviewed Changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

File	Description
kafka/adminapi_test.go	Added unit test to validate log event delivery.
kafka/adminapi.go	Added support for log channel configuration and termination.

Comments suppressed due to low confidence (1)

kafka/adminapi.go:3733

[nitpick] Consider using a channel of type struct{} for termination signals instead of bool to better convey that only a signal is needed.

	// before we do anything with the configuration, create a copy such that

Copilot · 2025-06-27T10:50:32Z

kafka/adminapi_test.go

+		{"INIT", "librdkafka"}: false,
+	}
+
+	go func() {


[nitpick] Consider adding an explicit termination condition in the goroutine reading from logsChan to avoid potential goroutine leaks when no more log events are sent. For example, signal closure of logsChan when the test completes.

Implemented in acdb44b

milindl

Just one change, and add a CHANGELOG.md entry, keep the version of the entry as 2.12 for now

milindl · 2025-07-07T02:30:36Z

kafka/adminapi.go

@@ -2737,6 +2738,9 @@ func (a *AdminClient) Close() {
 		return
 	}

+	if a.adminTermChan != nil {


Add a Logs() method onto the AdminClient, too, which returns a.handle.logs, to be used in case "go.logs.channel.enable": true, but "go.logs.channel": nil, or unset.

After this, if handle.closeLogsChan is true, then also close the logs channel here.

Thanks @milindl for review
Addressed feedback in commit 7245731

sonarqube-confluent · 2025-07-11T13:41:54Z

Analysis Details

3 Issues

0 Bugs
0 Vulnerabilities
3 Code Smells

Coverage and Duplications

93.30% Coverage (55.60% Estimated after merge)
No duplication information (0.10% Estimated after merge)

Project ID: confluent-kafka-go

View in SonarQube

vinicius91 · 2025-08-04T10:43:59Z

kafka/adminapi_test.go

@@ -1336,3 +1336,135 @@ func TestAdminAPIs(t *testing.T) {

 	a.Close()
 }
+
+func TestAdminClientLog(t *testing.T) {


Maybe something like this to reduce the Cognitive Complexity

Suggested change

func TestAdminClientLog(t *testing.T) {

func TestAdminClientLog(t *testing.T) {

logsChan := make(chan LogEvent, 100)

admin, err := NewAdminClient(&ConfigMap{

"debug": "all",

"bootstrap.servers": "localhost:65533", // Unreachable to generate logs

"go.logs.channel.enable": true,

"go.logs.channel": logsChan,

})

if err != nil {

t.Fatalf("Failed to create AdminClient: %v", err)

}

defer func() {

admin.Close()

close(logsChan)

}()

// Verify that Logs() method returns the correct channel

if admin.Logs() != logsChan {

t.Fatalf("Expected admin.Logs() %v == logsChan %v", admin.Logs(), logsChan)

}

// Define expected logs to validate

expectedLogs := []struct {

tag string

message string

found bool

}{

{"INIT", "librdkafka", false},

}

go func() {

for log := range logsChan {

t.Log(log.String())

// Check each expected log

for i := range expectedLogs {

if !expectedLogs[i].found && log.Tag == expectedLogs[i].tag &&

strings.Contains(log.Message, expectedLogs[i].message) {

expectedLogs[i].found = true

}

}

}

}()

<-time.After(time.Second * 5)

// Validate all expected logs were found

for _, expected := range expectedLogs {

if !expected.found {

t.Errorf("Expected to find log with tag `%s' and message containing `%s', but didn't find any.",

expected.tag, expected.message)

}

}

}

vinicius91 · 2025-08-04T10:46:57Z

kafka/adminapi_test.go

+	}
+}
+
+func TestAdminClientLogWithoutChannel(t *testing.T) {


Similar to the other comment

Suggested change

func TestAdminClientLogWithoutChannel(t *testing.T) {

func TestAdminClientLogWithoutChannel(t *testing.T) {

admin, err := NewAdminClient(&ConfigMap{

"debug": "all",

"bootstrap.servers": "localhost:65533", // Unreachable to generate logs

"go.logs.channel.enable": true,

// Note: no "go.logs.channel" specified

})

if err != nil {

t.Fatalf("Failed to create AdminClient: %v", err)

}

defer admin.Close()

// Verify that Logs() method returns a channel when enabled but no channel provided

logsChan := admin.Logs()

if logsChan == nil {

t.Fatalf("Expected admin.Logs() to return a channel, got nil")

}

// Define expected logs to validate

expectedLogs := []struct {

tag string

message string

found bool

}{

{"INIT", "librdkafka", false},

}

go func() {

for log := range logsChan {

t.Log(log.String())

// Check each expected log

for i := range expectedLogs {

if !expectedLogs[i].found && log.Tag == expectedLogs[i].tag &&

strings.Contains(log.Message, expectedLogs[i].message) {

expectedLogs[i].found = true

}

}

}

}()

<-time.After(time.Second * 5)

// Validate all expected logs were found

for _, expected := range expectedLogs {

if !expected.found {

t.Errorf("Expected to find log with tag `%s' and message containing `%s', but didn't find any.",

expected.tag, expected.message)

}

}

}

Copilot AI review requested due to automatic review settings June 27, 2025 10:34

k-raina requested review from a team as code owners June 27, 2025 10:34

Copilot AI reviewed Jun 27, 2025

View reviewed changes

k-raina self-assigned this Jun 27, 2025

milindl requested changes Jul 7, 2025

View reviewed changes

k-raina added 3 commits July 11, 2025 18:46

Issue-1119 : Add log channel to admin client

563fe37

Add copilot feedback

672a68f

Address feedback

f9051be

airlock-confluentinc bot force-pushed the issue-1119-add-logchannel-adminClient branch from 7245731 to f9051be Compare July 11, 2025 13:19

k-raina requested a review from milindl July 11, 2025 13:25

vinicius91 reviewed Aug 4, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Add log channel to admin client #1448

Add log channel to admin client #1448

Uh oh!

k-raina commented Jun 27, 2025

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI Jun 27, 2025

Uh oh!

k-raina Jun 27, 2025

Uh oh!

milindl left a comment

Uh oh!

milindl Jul 7, 2025

Uh oh!

k-raina Jul 11, 2025

Uh oh!

sonarqube-confluent bot commented Jul 11, 2025

Uh oh!

vinicius91 Aug 4, 2025

Uh oh!

vinicius91 Aug 4, 2025

Uh oh!

Uh oh!

-func TestAdminClientLog(t *testing.T) {
+func TestAdminClientLog(t *testing.T) {
+	logsChan := make(chan LogEvent, 100)
+	admin, err := NewAdminClient(&ConfigMap{
+		"debug":                  "all",
+		"bootstrap.servers":      "localhost:65533", // Unreachable to generate logs
+		"go.logs.channel.enable": true,
+		"go.logs.channel":        logsChan,
+	})
+	if err != nil {
+		t.Fatalf("Failed to create AdminClient: %v", err)
+	}
+	defer func() {
+		admin.Close()
+		close(logsChan)
+	}()
+	// Verify that Logs() method returns the correct channel
+	if admin.Logs() != logsChan {
+		t.Fatalf("Expected admin.Logs() %v == logsChan %v", admin.Logs(), logsChan)
+	}
+	// Define expected logs to validate
+	expectedLogs := []struct {
+		tag     string
+		message string
+		found   bool
+	}{
+		{"INIT", "librdkafka", false},
+	}
+	go func() {
+		for log := range logsChan {
+			t.Log(log.String())
+			// Check each expected log
+			for i := range expectedLogs {
+				if !expectedLogs[i].found && log.Tag == expectedLogs[i].tag &&
+					strings.Contains(log.Message, expectedLogs[i].message) {
+					expectedLogs[i].found = true
+				}
+			}
+		}
+	}()
+	<-time.After(time.Second * 5)
+	// Validate all expected logs were found
+	for _, expected := range expectedLogs {
+		if !expected.found {
+			t.Errorf("Expected to find log with tag `%s' and message containing `%s', but didn't find any.",
+				expected.tag, expected.message)
+		}
+	}
+}

Add log channel to admin client #1448

Are you sure you want to change the base?

Add log channel to admin client #1448

Uh oh!

Conversation

k-raina commented Jun 27, 2025

Summary

Motivation

Changes

Test

Unit Test

Manual test

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Copilot AI Jun 27, 2025

Choose a reason for hiding this comment

Uh oh!

k-raina Jun 27, 2025

Choose a reason for hiding this comment

Uh oh!

milindl left a comment

Choose a reason for hiding this comment

Uh oh!

milindl Jul 7, 2025

Choose a reason for hiding this comment

Uh oh!

k-raina Jul 11, 2025

Choose a reason for hiding this comment

Uh oh!

sonarqube-confluent bot commented Jul 11, 2025

Analysis Details

3 Issues

Coverage and Duplications

Uh oh!

vinicius91 Aug 4, 2025

Choose a reason for hiding this comment

Uh oh!

vinicius91 Aug 4, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!