Skip to content
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

Finish generating LogDirectory before booting up any executable #41

Merged
merged 15 commits into from
Jan 7, 2025
Merged

Finish generating LogDirectory before booting up any executable #41

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
9a53f5b
refactor: streamline logging and error handling in testBindToPort fun…
ryan-gang Jan 6, 2025
f3e1359
refactor: improve logging consistency and error handling in testHardc…
ryan-gang Jan 6, 2025
b279a72
refactor: enhance logging and error handling in testCorrelationId fun…
ryan-gang Jan 6, 2025
ed2ce5f
refactor: unify logging and error handling in testAPIVersionErrorCase…
ryan-gang Jan 6, 2025
6251c74
refactor: unify logging and error handling in testAPIVersion function
ryan-gang Jan 6, 2025
689831d
refactor: enhance logging and error handling in testSequentialRequest…
ryan-gang Jan 6, 2025
c8e65fd
refactor: unify logging and error handling in testConcurrentRequests …
ryan-gang Jan 6, 2025
51bc5ca
refactor: unify logging and error handling in testAPIVersionWithFetch…
ryan-gang Jan 6, 2025
0fb2a17
refactor: unify logging and error handling in testFetchWithNoTopics f…
ryan-gang Jan 6, 2025
aa57862
refactor: unify logging and error handling in testFetchWithUnknownTop…
ryan-gang Jan 6, 2025
13f3599
refactor: enhance logging and error handling in testFetchNoMessages f…
ryan-gang Jan 6, 2025
179d40b
refactor: unify logging and error handling in multiple test functions
ryan-gang Jan 6, 2025
cd472c9
chore: fix typos
ryan-gang Jan 6, 2025
5e6f1e2
refactor: remove unused GetDescribeTopicPartition function
ryan-gang Jan 6, 2025
df9178b
test: update fixtures
ryan-gang Jan 6, 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
11 changes: 5 additions & 6 deletions internal/stage1.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,14 +10,13 @@ import (

func testBindToPort(stageHarness *test_case_harness.TestCaseHarness) error {
b := kafka_executable.NewKafkaExecutable(stageHarness)
if err := b.Run(); err != nil {
err := serializer.GenerateLogDirs(logger.GetQuietLogger(""), true)
if err != nil {
return err
}

quietLogger := logger.GetQuietLogger("")
logger := stageHarness.Logger
err := serializer.GenerateLogDirs(quietLogger, true)
if err != nil {
stageLogger := stageHarness.Logger
if err := b.Run(); err != nil {
return err
}

Expand All @@ -26,5 +25,5 @@ func testBindToPort(stageHarness *test_case_harness.TestCaseHarness) error {
Retries: 15,
}

return bindTestCase.Run(b, logger)
return bindTestCase.Run(b, stageLogger)
}
35 changes: 18 additions & 17 deletions internal/stage2.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,22 +16,23 @@ import (

func testHardcodedCorrelationId(stageHarness *test_case_harness.TestCaseHarness) error {
b := kafka_executable.NewKafkaExecutable(stageHarness)
if err := b.Run(); err != nil {
err := serializer.GenerateLogDirs(logger.GetQuietLogger(""), true)
if err != nil {
return err
}

quietLogger := logger.GetQuietLogger("")
logger := stageHarness.Logger
err := serializer.GenerateLogDirs(quietLogger, true)
if err != nil {
stageLogger := stageHarness.Logger
if err := b.Run(); err != nil {
return err
}

broker := protocol.NewBroker("localhost:9092")
if err := broker.ConnectWithRetries(b, logger); err != nil {
if err := broker.ConnectWithRetries(b, stageLogger); err != nil {
return err
}
defer broker.Close()
defer func(broker *protocol.Broker) {
_ = broker.Close()
}(broker)

correlationId := 7

Expand All @@ -50,8 +51,8 @@ func testHardcodedCorrelationId(stageHarness *test_case_harness.TestCaseHarness)
}

message := kafkaapi.EncodeApiVersionsRequest(&request)
logger.Infof("Sending \"ApiVersions\" (version: %v) request (Correlation id: %v)", request.Header.ApiVersion, request.Header.CorrelationId)
logger.Debugf("Hexdump of sent \"ApiVersions\" request: \n%v\n", GetFormattedHexdump(message))
stageLogger.Infof("Sending \"ApiVersions\" (version: %v) request (Correlation id: %v)", request.Header.ApiVersion, request.Header.CorrelationId)
stageLogger.Debugf("Hexdump of sent \"ApiVersions\" request: \n%v\n", GetFormattedHexdump(message))

err = broker.Send(message)
if err != nil {
Expand All @@ -61,13 +62,13 @@ func testHardcodedCorrelationId(stageHarness *test_case_harness.TestCaseHarness)
if err != nil {
return err
}
logger.Debugf("Hexdump of received \"ApiVersions\" response: \n%v\n", GetFormattedHexdump(response))
stageLogger.Debugf("Hexdump of received \"ApiVersions\" response: \n%v\n", GetFormattedHexdump(response))

decoder := realdecoder.RealDecoder{}
decoder.Init(response)
logger.UpdateSecondaryPrefix("Decoder")
stageLogger.UpdateSecondaryPrefix("Decoder")

logger.Debugf("- .Response")
stageLogger.Debugf("- .Response")
messageLength, err := decoder.GetInt32()
if err != nil {
if decodingErr, ok := err.(*errors.PacketDecodingError); ok {
Expand All @@ -76,9 +77,9 @@ func testHardcodedCorrelationId(stageHarness *test_case_harness.TestCaseHarness)
}
return err
}
protocol.LogWithIndentation(logger, 1, "- .message_length (%d)", messageLength)
protocol.LogWithIndentation(stageLogger, 1, "- .message_length (%d)", messageLength)

logger.Debugf("- .ResponseHeader")
stageLogger.Debugf("- .ResponseHeader")
responseCorrelationId, err := decoder.GetInt32()
if err != nil {
if decodingErr, ok := err.(*errors.PacketDecodingError); ok {
Expand All @@ -87,14 +88,14 @@ func testHardcodedCorrelationId(stageHarness *test_case_harness.TestCaseHarness)
}
return err
}
protocol.LogWithIndentation(logger, 1, "- .correlation_id (%d)", responseCorrelationId)
logger.ResetSecondaryPrefix()
protocol.LogWithIndentation(stageLogger, 1, "- .correlation_id (%d)", responseCorrelationId)
stageLogger.ResetSecondaryPrefix()

if responseCorrelationId != int32(correlationId) {
return fmt.Errorf("Expected Correlation ID to be %v, got %v", int32(correlationId), responseCorrelationId)
}

logger.Successf("✓ Correlation ID: %v", responseCorrelationId)
stageLogger.Successf("✓ Correlation ID: %v", responseCorrelationId)

return nil
}
37 changes: 19 additions & 18 deletions internal/stage3.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ package internal

import (
"fmt"
"github.com/codecrafters-io/tester-utils/logger"

realdecoder "github.com/codecrafters-io/kafka-tester/protocol/decoder"

Expand All @@ -10,28 +11,28 @@ import (
kafkaapi "github.com/codecrafters-io/kafka-tester/protocol/api"
"github.com/codecrafters-io/kafka-tester/protocol/errors"
"github.com/codecrafters-io/kafka-tester/protocol/serializer"
"github.com/codecrafters-io/tester-utils/logger"
"github.com/codecrafters-io/tester-utils/test_case_harness"
)

func testCorrelationId(stageHarness *test_case_harness.TestCaseHarness) error {
b := kafka_executable.NewKafkaExecutable(stageHarness)
if err := b.Run(); err != nil {
err := serializer.GenerateLogDirs(logger.GetQuietLogger(""), true)
if err != nil {
return err
}

quietLogger := logger.GetQuietLogger("")
logger := stageHarness.Logger
err := serializer.GenerateLogDirs(quietLogger, true)
if err != nil {
stageLogger := stageHarness.Logger
if err := b.Run(); err != nil {
return err
}

broker := protocol.NewBroker("localhost:9092")
if err := broker.ConnectWithRetries(b, logger); err != nil {
if err := broker.ConnectWithRetries(b, stageLogger); err != nil {
return err
}
defer broker.Close()
defer func(broker *protocol.Broker) {
_ = broker.Close()
}(broker)

correlationId := getRandomCorrelationId()

Expand All @@ -50,8 +51,8 @@ func testCorrelationId(stageHarness *test_case_harness.TestCaseHarness) error {
}

message := kafkaapi.EncodeApiVersionsRequest(&request)
logger.Infof("Sending \"ApiVersions\" (version: %v) request (Correlation id: %v)", request.Header.ApiVersion, request.Header.CorrelationId)
logger.Debugf("Hexdump of sent \"ApiVersions\" request: \n%v\n", GetFormattedHexdump(message))
stageLogger.Infof("Sending \"ApiVersions\" (version: %v) request (Correlation id: %v)", request.Header.ApiVersion, request.Header.CorrelationId)
stageLogger.Debugf("Hexdump of sent \"ApiVersions\" request: \n%v\n", GetFormattedHexdump(message))

err = broker.Send(message)
if err != nil {
Expand All @@ -61,13 +62,13 @@ func testCorrelationId(stageHarness *test_case_harness.TestCaseHarness) error {
if err != nil {
return err
}
logger.Debugf("Hexdump of received \"ApiVersions\" response: \n%v\n", GetFormattedHexdump(response))
stageLogger.Debugf("Hexdump of received \"ApiVersions\" response: \n%v\n", GetFormattedHexdump(response))

decoder := realdecoder.RealDecoder{}
decoder.Init(response)
logger.UpdateSecondaryPrefix("Decoder")
stageLogger.UpdateSecondaryPrefix("Decoder")

logger.Debugf("- .Response")
stageLogger.Debugf("- .Response")
messageLength, err := decoder.GetInt32()
if err != nil {
if decodingErr, ok := err.(*errors.PacketDecodingError); ok {
Expand All @@ -76,9 +77,9 @@ func testCorrelationId(stageHarness *test_case_harness.TestCaseHarness) error {
}
return err
}
protocol.LogWithIndentation(logger, 1, "- .message_length (%d)", messageLength)
protocol.LogWithIndentation(stageLogger, 1, "- .message_length (%d)", messageLength)

logger.Debugf("- .ResponseHeader")
stageLogger.Debugf("- .ResponseHeader")
responseCorrelationId, err := decoder.GetInt32()
if err != nil {
if decodingErr, ok := err.(*errors.PacketDecodingError); ok {
Expand All @@ -87,14 +88,14 @@ func testCorrelationId(stageHarness *test_case_harness.TestCaseHarness) error {
}
return err
}
protocol.LogWithIndentation(logger, 1, "- .correlation_id (%d)", responseCorrelationId)
logger.ResetSecondaryPrefix()
protocol.LogWithIndentation(stageLogger, 1, "- .correlation_id (%d)", responseCorrelationId)
stageLogger.ResetSecondaryPrefix()

if responseCorrelationId != correlationId {
return fmt.Errorf("Expected Correlation ID to be %v, got %v", correlationId, responseCorrelationId)
}

logger.Successf("✓ Correlation ID: %v", responseCorrelationId)
stageLogger.Successf("✓ Correlation ID: %v", responseCorrelationId)

return nil
}
41 changes: 21 additions & 20 deletions internal/stage4.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,38 +2,39 @@ package internal

import (
"fmt"
"github.com/codecrafters-io/tester-utils/logger"

"github.com/codecrafters-io/kafka-tester/internal/kafka_executable"
"github.com/codecrafters-io/kafka-tester/protocol"
kafkaapi "github.com/codecrafters-io/kafka-tester/protocol/api"
realdecoder "github.com/codecrafters-io/kafka-tester/protocol/decoder"
"github.com/codecrafters-io/kafka-tester/protocol/errors"
"github.com/codecrafters-io/kafka-tester/protocol/serializer"
"github.com/codecrafters-io/tester-utils/logger"
"github.com/codecrafters-io/tester-utils/test_case_harness"
)

func testAPIVersionErrorCase(stageHarness *test_case_harness.TestCaseHarness) error {
b := kafka_executable.NewKafkaExecutable(stageHarness)
if err := b.Run(); err != nil {
err := serializer.GenerateLogDirs(logger.GetQuietLogger(""), true)
if err != nil {
return err
}

quietLogger := logger.GetQuietLogger("")
logger := stageHarness.Logger
err := serializer.GenerateLogDirs(quietLogger, true)
if err != nil {
stageLogger := stageHarness.Logger
if err := b.Run(); err != nil {
return err
}

correlationId := getRandomCorrelationId()
apiVersion := getInvalidAPIVersion()

broker := protocol.NewBroker("localhost:9092")
if err := broker.ConnectWithRetries(b, logger); err != nil {
if err := broker.ConnectWithRetries(b, stageLogger); err != nil {
return err
}
defer broker.Close()
defer func(broker *protocol.Broker) {
_ = broker.Close()
}(broker)

request := kafkaapi.ApiVersionsRequest{
Header: kafkaapi.RequestHeader{
Expand All @@ -50,8 +51,8 @@ func testAPIVersionErrorCase(stageHarness *test_case_harness.TestCaseHarness) er
}

message := kafkaapi.EncodeApiVersionsRequest(&request)
logger.Infof("Sending \"ApiVersions\" (version: %v) request (Correlation id: %v)", request.Header.ApiVersion, request.Header.CorrelationId)
logger.Debugf("Hexdump of sent \"ApiVersions\" request: \n%v\n", GetFormattedHexdump(message))
stageLogger.Infof("Sending \"ApiVersions\" (version: %v) request (Correlation id: %v)", request.Header.ApiVersion, request.Header.CorrelationId)
stageLogger.Debugf("Hexdump of sent \"ApiVersions\" request: \n%v\n", GetFormattedHexdump(message))

err = broker.Send(message)
if err != nil {
Expand All @@ -61,13 +62,13 @@ func testAPIVersionErrorCase(stageHarness *test_case_harness.TestCaseHarness) er
if err != nil {
return err
}
logger.Debugf("Hexdump of received \"ApiVersions\" response: \n%v\n", GetFormattedHexdump(response))
stageLogger.Debugf("Hexdump of received \"ApiVersions\" response: \n%v\n", GetFormattedHexdump(response))

decoder := realdecoder.RealDecoder{}
decoder.Init(response)
logger.UpdateSecondaryPrefix("Decoder")
stageLogger.UpdateSecondaryPrefix("Decoder")

logger.Debugf("- .Response")
stageLogger.Debugf("- .Response")
messageLength, err := decoder.GetInt32()
if err != nil {
if decodingErr, ok := err.(*errors.PacketDecodingError); ok {
Expand All @@ -76,9 +77,9 @@ func testAPIVersionErrorCase(stageHarness *test_case_harness.TestCaseHarness) er
}
return err
}
protocol.LogWithIndentation(logger, 1, "- .message_length (%d)", messageLength)
protocol.LogWithIndentation(stageLogger, 1, "- .message_length (%d)", messageLength)

logger.Debugf("- .ResponseHeader")
stageLogger.Debugf("- .ResponseHeader")
responseCorrelationId, err := decoder.GetInt32()
if err != nil {
if decodingErr, ok := err.(*errors.PacketDecodingError); ok {
Expand All @@ -87,7 +88,7 @@ func testAPIVersionErrorCase(stageHarness *test_case_harness.TestCaseHarness) er
}
return err
}
protocol.LogWithIndentation(logger, 1, "- .correlation_id (%d)", responseCorrelationId)
protocol.LogWithIndentation(stageLogger, 1, "- .correlation_id (%d)", responseCorrelationId)

errorCode, err := decoder.GetInt16()
if err != nil {
Expand All @@ -97,20 +98,20 @@ func testAPIVersionErrorCase(stageHarness *test_case_harness.TestCaseHarness) er
}
return err
}
protocol.LogWithIndentation(logger, 1, "- .error_code (%d)", errorCode)
logger.ResetSecondaryPrefix()
protocol.LogWithIndentation(stageLogger, 1, "- .error_code (%d)", errorCode)
stageLogger.ResetSecondaryPrefix()

if responseCorrelationId != correlationId {
return fmt.Errorf("Expected Correlation ID to be %v, got %v", correlationId, responseCorrelationId)
}

logger.Successf("✓ Correlation ID: %v", responseCorrelationId)
stageLogger.Successf("✓ Correlation ID: %v", responseCorrelationId)

if errorCode != 35 {
return fmt.Errorf("Expected Error code to be 35, got %v", errorCode)
}

logger.Successf("✓ Error code: 35 (UNSUPPORTED_VERSION)")
stageLogger.Successf("✓ Error code: 35 (UNSUPPORTED_VERSION)")

return nil
}
Loading
Loading