From 75c807b586b0b83b041980ec1a280f292f3d2e3a Mon Sep 17 00:00:00 2001 From: Lars Lehtonen Date: Wed, 13 May 2026 13:59:24 -0700 Subject: [PATCH] chore(weed/mq/kafka/protocol): remove unused functions and variables (#9488) --- .../kafka/protocol/consumer_coordination.go | 26 ------------------- weed/mq/kafka/protocol/handler.go | 2 -- .../protocol/offset_fetch_pattern_test.go | 14 +--------- .../response_validation_example_test.go | 10 ------- 4 files changed, 1 insertion(+), 51 deletions(-) diff --git a/weed/mq/kafka/protocol/consumer_coordination.go b/weed/mq/kafka/protocol/consumer_coordination.go index 9e3db02aa..ac6a1486b 100644 --- a/weed/mq/kafka/protocol/consumer_coordination.go +++ b/weed/mq/kafka/protocol/consumer_coordination.go @@ -519,23 +519,6 @@ func (h *Handler) parseLeaveGroupRequest(data []byte, apiVersion uint16) (*Leave return req, nil } -func (h *Handler) buildHeartbeatResponse(response HeartbeatResponse) []byte { - result := make([]byte, 0, 12) - - // NOTE: Correlation ID is handled by writeResponseWithCorrelationID - // Do NOT include it in the response body - - // Error code (2 bytes) - errorCodeBytes := make([]byte, 2) - binary.BigEndian.PutUint16(errorCodeBytes, uint16(response.ErrorCode)) - result = append(result, errorCodeBytes...) - - // Throttle time (4 bytes, 0 = no throttling) - result = append(result, 0, 0, 0, 0) - - return result -} - func (h *Handler) buildHeartbeatResponseV(response HeartbeatResponse, apiVersion uint16) []byte { isFlexible := IsFlexibleVersion(12, apiVersion) // Heartbeat API key = 12 result := make([]byte, 0, 16) @@ -717,15 +700,6 @@ func (h *Handler) buildLeaveGroupFullResponse(response LeaveGroupResponse) []byt return result } -func (h *Handler) buildHeartbeatErrorResponse(correlationID uint32, errorCode int16) []byte { - response := HeartbeatResponse{ - CorrelationID: correlationID, - ErrorCode: errorCode, - } - - return h.buildHeartbeatResponse(response) -} - func (h *Handler) buildHeartbeatErrorResponseV(correlationID uint32, errorCode int16, apiVersion uint16) []byte { response := HeartbeatResponse{ CorrelationID: correlationID, diff --git a/weed/mq/kafka/protocol/handler.go b/weed/mq/kafka/protocol/handler.go index 8525c88ac..0e726fd1d 100644 --- a/weed/mq/kafka/protocol/handler.go +++ b/weed/mq/kafka/protocol/handler.go @@ -286,8 +286,6 @@ type Handler struct { inferredRecordTypes map[string]*schema_pb.RecordType inferredRecordTypesMu sync.RWMutex - filerClient filer_pb.SeaweedFilerClient - // SMQ broker addresses discovered from masters for Metadata responses smqBrokerAddresses []string diff --git a/weed/mq/kafka/protocol/offset_fetch_pattern_test.go b/weed/mq/kafka/protocol/offset_fetch_pattern_test.go index 544e0c9f6..fdd168edf 100644 --- a/weed/mq/kafka/protocol/offset_fetch_pattern_test.go +++ b/weed/mq/kafka/protocol/offset_fetch_pattern_test.go @@ -98,16 +98,6 @@ func TestOffsetFetchAfterCommit(t *testing.T) { t.Skip("Integration test - requires mock broker setup") t.Run("FetchAfterCommit", func(t *testing.T) { - type FetchRequest struct { - partition int32 - offset int64 - } - - type FetchResponse struct { - records []byte - nextOffset int64 - } - // Simulate: Commit offset 163, then fetch offset 164 committedOffset := int64(163) nextFetchOffset := committedOffset + 1 @@ -213,10 +203,8 @@ func TestLongPollWithOffsetCommit(t *testing.T) { // Critical: long-poll duration should NOT be reported as throttleTimeMs // This was bug 8969b4509 - const maxWaitTime = 5 * time.Second - // Simulate long-poll wait (no data available) - time.Sleep(100 * time.Millisecond) // Broker waits up to maxWaitTime + time.Sleep(100 * time.Millisecond) // throttleTimeMs should be 0 (NOT elapsed duration!) throttleTimeMs := int32(0) // CORRECT diff --git a/weed/mq/kafka/protocol/response_validation_example_test.go b/weed/mq/kafka/protocol/response_validation_example_test.go index a69c03f4f..3d07de756 100644 --- a/weed/mq/kafka/protocol/response_validation_example_test.go +++ b/weed/mq/kafka/protocol/response_validation_example_test.go @@ -100,16 +100,6 @@ func TestCurrentTestingApproach(t *testing.T) { t.Log(" This is why JoinGroup issue wasn't caught by unit tests") } -// parseCompactArray is a helper that would be needed for field-level testing -func parseCompactArray(data []byte) int { - // Compact array encoding: varint length (length+1 for non-null, 0 for null) - length := int(data[0]) - if length == 0 { - return -1 // null - } - return length - 1 // actual length -} - // Example of a REAL field-level test we could write func TestMetadataResponseHasBrokers(t *testing.T) { t.Skip("Example of what a real field-level test would look like")