fix(stage_p2): use stage logger for GenerateLogDirs to ensure consistent logging

feat: add testProduce2 to validate produce request error handling
fix: simplify deferred client close by removing wrapper function

Author: ryan-gang

internal/stage_p2.go

@@ -0,0 +1,62 @@
+package internal
+
+import (
+	"github.com/codecrafters-io/kafka-tester/internal/assertions"
+	"github.com/codecrafters-io/kafka-tester/internal/kafka_executable"
+	"github.com/codecrafters-io/kafka-tester/protocol/builder"
+	"github.com/codecrafters-io/kafka-tester/protocol/common"
+	"github.com/codecrafters-io/kafka-tester/protocol/kafka_client"
+	"github.com/codecrafters-io/kafka-tester/protocol/serializer"
+	"github.com/codecrafters-io/tester-utils/test_case_harness"
+)
+
+func testProduce2(stageHarness *test_case_harness.TestCaseHarness) error {
+	b := kafka_executable.NewKafkaExecutable(stageHarness)
+	stageLogger := stageHarness.Logger
+	err := serializer.GenerateLogDirs(stageLogger, []string{})
+	if err != nil {
+		return err
+	}
+
+	if err := b.Run(); err != nil {
+		return err
+	}
+
+	correlationId := getRandomCorrelationId()
+
+	client := kafka_client.NewClient("localhost:9092")
+	if err := client.ConnectWithRetries(b, stageLogger); err != nil {
+		return err
+	}
+	defer client.Close()
+
+	recordBatch := builder.NewRecordBatchBuilder().
+		AddStringRecord(common.MESSAGE1).
+		Build()
+
+	request := builder.NewProduceRequestBuilder().
+		AddRecordBatch(common.TOPIC_UNKOWN_NAME, 0, recordBatch).
+		WithCorrelationId(correlationId).
+		Build()
+
+	rawResponse, err := client.SendAndReceive(request, stageLogger)
+	if err != nil {
+		return err
+	}
+
+	actualResponse := builder.NewEmptyProduceResponse()
+	if err := actualResponse.Decode(rawResponse.Payload, stageLogger); err != nil {
+		return err
+	}
+
+	expectedResponse := builder.NewProduceResponseBuilder().
+		AddErrorPartitionResponse(common.TOPIC_UNKOWN_NAME, 0, 3).
+		WithCorrelationId(correlationId).
+		Build()
+
+	if err = assertions.NewProduceResponseAssertion(actualResponse, expectedResponse, stageLogger).Run(stageLogger); err != nil {
+		return err
+	}
+
+	return nil
+}