[exporter/kafka] franz-go: Fix underreported kafka_exporter_write_latency metric (open-telemetry#43859)

#### Description
franz-go: Fix underreported kafka_exporter_write_latency metric by using
OnBrokerE2E franz-go hook that takes response read time into account

<!-- Issue number (e.g. open-telemetry#1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Fixes open-telemetry#43803

<!--Describe what testing was performed and which tests were added.-->
#### Testing

<!--Describe the documentation added.-->
#### Documentation

<!--Please delete paragraphs that you did not use before submitting.-->

Author: carsonip

.chloggen/kafkaexporter_franzgo-write-metric.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. receiver/filelog)
+component: exporter/kafka
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "franz-go: Fix underreported kafka_exporter_write_latency metric"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [43803]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

exporter/kafkaexporter/internal/kafkaclient/franzgo_metrics.go

@@ -26,6 +26,8 @@ func NewFranzProducerMetrics(tb *metadata.TelemetryBuilder) FranzProducerMetrics
 	return FranzProducerMetrics{tb: tb}
 }
 
+var _ kgo.HookBrokerConnect = FranzProducerMetrics{}
+
 func (fpm FranzProducerMetrics) OnBrokerConnect(meta kgo.BrokerMetadata, _ time.Duration, _ net.Conn, err error) {
 	outcome := "success"
 	if err != nil {
@@ -41,6 +43,8 @@ func (fpm FranzProducerMetrics) OnBrokerConnect(meta kgo.BrokerMetadata, _ time.
 	)
 }
 
+var _ kgo.HookBrokerDisconnect = FranzProducerMetrics{}
+
 func (fpm FranzProducerMetrics) OnBrokerDisconnect(meta kgo.BrokerMetadata, _ net.Conn) {
 	fpm.tb.KafkaBrokerClosed.Add(
 		context.Background(),
@@ -51,6 +55,8 @@ func (fpm FranzProducerMetrics) OnBrokerDisconnect(meta kgo.BrokerMetadata, _ ne
 	)
 }
 
+var _ kgo.HookBrokerThrottle = FranzProducerMetrics{}
+
 func (fpm FranzProducerMetrics) OnBrokerThrottle(meta kgo.BrokerMetadata, throttleInterval time.Duration, _ bool) {
 	// KafkaBrokerThrottlingDuration is deprecated in favor of KafkaBrokerThrottlingLatency.
 	fpm.tb.KafkaBrokerThrottlingDuration.Record(
@@ -69,30 +75,34 @@ func (fpm FranzProducerMetrics) OnBrokerThrottle(meta kgo.BrokerMetadata, thrott
 	)
 }
 
-func (fpm FranzProducerMetrics) OnBrokerWrite(meta kgo.BrokerMetadata, _ int16, _ int, writeWait, timeToWrite time.Duration, err error) {
+var _ kgo.HookBrokerE2E = FranzProducerMetrics{}
+
+func (fpm FranzProducerMetrics) OnBrokerE2E(meta kgo.BrokerMetadata, _ int16, e2e kgo.BrokerE2E) {
 	outcome := "success"
-	if err != nil {
+	if e2e.Err() != nil {
 		outcome = "failure"
 	}
 	// KafkaExporterLatency is deprecated in favor of KafkaExporterWriteLatency.
 	fpm.tb.KafkaExporterLatency.Record(
 		context.Background(),
-		writeWait.Milliseconds()+timeToWrite.Milliseconds(),
+		e2e.DurationE2E().Milliseconds()+e2e.WriteWait.Milliseconds(),
 		metric.WithAttributes(
 			attribute.String("node_id", kgo.NodeName(meta.NodeID)),
 			attribute.String("outcome", outcome),
 		),
 	)
 	fpm.tb.KafkaExporterWriteLatency.Record(
 		context.Background(),
-		writeWait.Seconds()+timeToWrite.Seconds(),
+		e2e.DurationE2E().Seconds()+e2e.WriteWait.Seconds(),
 		metric.WithAttributes(
 			attribute.String("node_id", kgo.NodeName(meta.NodeID)),
 			attribute.String("outcome", outcome),
 		),
 	)
 }
 
+var _ kgo.HookProduceBatchWritten = FranzProducerMetrics{}
+
 // OnProduceBatchWritten is called when a batch has been produced.
 // https://pkg.go.dev/github.com/twmb/franz-go/pkg/kgo#HookProduceBatchWritten
 func (fpm FranzProducerMetrics) OnProduceBatchWritten(meta kgo.BrokerMetadata, topic string, partition int32, m kgo.ProduceBatchMetrics) {
@@ -126,6 +136,8 @@ func (fpm FranzProducerMetrics) OnProduceBatchWritten(meta kgo.BrokerMetadata, t
 	)
 }
 
+var _ kgo.HookProduceRecordUnbuffered = FranzProducerMetrics{}
+
 // OnProduceRecordUnbuffered records the number of produced messages that were
 // not produced due to errors. The successfully produced records is recorded by
 // `OnProduceBatchWritten`.

exporter/kafkaexporter/internal/kafkaclient/franzgo_metrics_test.go

@@ -79,14 +79,28 @@ func TestFranzProducerMetrics(t *testing.T) {
 			metricdatatest.IgnoreTimestamp(),
 		)
 	})
-	t.Run("should report the metrics when OnBrokerWrite hook is called", func(t *testing.T) {
+	t.Run("should report the metrics when OnBrokerE2E hook is called", func(t *testing.T) {
 		testTel := componenttest.NewTelemetry()
 		tb, err := metadata.NewTelemetryBuilder(testTel.NewTelemetrySettings())
 		require.NoError(t, err)
 		defer tb.Shutdown()
 		fpm := NewFranzProducerMetrics(tb)
-		fpm.OnBrokerWrite(kgo.BrokerMetadata{NodeID: 1}, 0, 0, time.Second/2, time.Second/2, nil)
-		fpm.OnBrokerWrite(kgo.BrokerMetadata{NodeID: 1}, 0, 0, 100*time.Second, 0, errors.New(""))
+		fpm.OnBrokerE2E(kgo.BrokerMetadata{NodeID: 1}, 0, kgo.BrokerE2E{
+			WriteWait:   time.Second / 4,
+			TimeToWrite: time.Second / 4,
+			ReadWait:    time.Second / 4,
+			TimeToRead:  time.Second / 4,
+			WriteErr:    nil,
+			ReadErr:     nil,
+		})
+		fpm.OnBrokerE2E(kgo.BrokerMetadata{NodeID: 1}, 0, kgo.BrokerE2E{
+			WriteWait:   time.Second * 25,
+			TimeToWrite: time.Second * 25,
+			ReadWait:    time.Second * 25,
+			TimeToRead:  time.Second * 25,
+			WriteErr:    errors.New(""),
+			ReadErr:     nil,
+		})
 		var rm metricdata.ResourceMetrics
 		err = testTel.Reader.Collect(t.Context(), &rm)
 		require.NoError(t, err)