From 628221faa319b68714ebc8380e38bf1933dbeda0 Mon Sep 17 00:00:00 2001 From: Dom Del Nano Date: Mon, 13 Jul 2026 07:45:26 -0700 Subject: [PATCH] Refresh kafka protocol parsing api versions Signed-off-by: Dom Del Nano --- bazel/container_images.bzl | 30 ++- .../protocol_inference/model/ruleset_basic.py | 6 +- .../socket_tracer/BUILD.bazel | 1 - .../bcc_bpf/protocol_inference.h | 4 +- .../socket_tracer/kafka_trace_bpf_test.cc | 246 +++++++----------- .../protocols/kafka/common/types.h | 144 +++++++--- .../protocols/kafka/decoder/fetch.cc | 27 +- .../protocols/kafka/decoder/fetch_test.cc | 77 ++++++ .../protocols/kafka/decoder/join_group.cc | 11 + .../protocols/kafka/decoder/metadata.cc | 4 +- .../protocols/kafka/decoder/packet_decoder.cc | 20 ++ .../protocols/kafka/decoder/packet_decoder.h | 6 + .../kafka/decoder/packet_decoder_test.cc | 25 ++ .../protocols/kafka/decoder/produce.cc | 18 +- .../protocols/kafka/decoder/produce_test.cc | 28 ++ .../protocols/kafka/opcodes/fetch.h | 15 ++ .../protocols/kafka/opcodes/produce.h | 10 + .../testing/container_images/BUILD.bazel | 10 +- .../container_images/kafka_container.h | 28 +- .../container_images/zookeeper_container.h | 45 ---- .../testing/containers/BUILD.bazel | 4 +- 21 files changed, 487 insertions(+), 272 deletions(-) delete mode 100644 src/stirling/source_connectors/socket_tracer/testing/container_images/zookeeper_container.h diff --git a/bazel/container_images.bzl b/bazel/container_images.bzl index ab81087c1d6..fe34dffe89a 100644 --- a/bazel/container_images.bzl +++ b/bazel/container_images.bzl @@ -41,6 +41,18 @@ def _container_image(name, digest, repository): repository = image_repo, ) +# Pulls an image directly from its upstream Docker Hub repository, bypassing the +# pixie-io registry mirror. Use this only for images that have not yet been mirrored +# (see scripts/regclient/regbot_deps.yaml). Prefer _container_image once the image is mirrored. +# TODO(ddelnano): Mirror the Kafka test images and switch these back to _container_image. +def _upstream_container_image(name, digest, repository): + container_pull( + name = name, + digest = digest, + registry = "index.docker.io", + repository = repository, + ) + def base_images(): # Based on alpine 3.15.9, using OpenResty 1.21.4.1 # https://hub.docker.com/layers/openresty/openresty/alpine-apk-amd64/images/sha256-2259f28de01f85c22e32b6964254a4551c54a1d554cd4b5f1615d7497e1a09ce?context=explore @@ -221,18 +233,20 @@ def stirling_test_images(): digest = "sha256:55521ffe36911fb4edeaeecb7f9219f9d2a09bc275530212b89e41ab78a7f16d", ) - # Kafka broker image, for testing. - _container_image( + # Kafka broker image (Confluent packaging), for testing. + # Tag: confluentinc/cp-kafka:7.8.0 (Kafka 3.8), linux/amd64. Runs in KRaft mode. + _upstream_container_image( name = "kafka_base_image", repository = "confluentinc/cp-kafka", - digest = "sha256:ee6e42ce4f79623c69cf758848de6761c74bf9712697fe68d96291a2b655ce7f", + digest = "sha256:59787f748cee60476bc6d0509bec28e8e5e4225dc96ef9f09866fbde341202e4", ) - # Zookeeper image for Kafka. - _container_image( - name = "zookeeper_base_image", - repository = "confluentinc/cp-zookeeper", - digest = "sha256:87314e87320abf190f0407bf1689f4827661fbb4d671a41cba62673b45b66bfa", + # Kafka broker image (Apache upstream packaging), for testing. + # Tag: apache/kafka:4.0.0 (Kafka 4.0), linux/amd64. Runs in KRaft mode. + _upstream_container_image( + name = "apache_kafka_base_image", + repository = "apache/kafka", + digest = "sha256:01b9a4030e54c6068e66eb3ba4cb82c0d89238629ef1c30d79b86036bf89b1b7", ) # Tag: node:12.3.1-stretch-slim diff --git a/src/stirling/protocol_inference/model/ruleset_basic.py b/src/stirling/protocol_inference/model/ruleset_basic.py index c168f7d3531..4466cb1b982 100644 --- a/src/stirling/protocol_inference/model/ruleset_basic.py +++ b/src/stirling/protocol_inference/model/ruleset_basic.py @@ -247,15 +247,15 @@ def infer_mysql_message(buf, count): # correlation_id => INT32 def infer_kafka_request(buf): # Note: The Number of Kafka APIs and their versions might change in the future. - kNumAPIs = 62 - kMaxAPIVersion = 12 + kNumAPIs = 94 + kMaxAPIVersion = 18 requestAPIKey = int.from_bytes(buf[4:6], byteorder="big") if requestAPIKey < 0 or requestAPIKey >= kNumAPIs: return MessageType.kUnknown requestAPIVersion = int.from_bytes(buf[6:8], byteorder="big") - if requestAPIVersion < 0 or requestAPIKey > kMaxAPIVersion: + if requestAPIVersion < 0 or requestAPIVersion > kMaxAPIVersion: return MessageType.kUnknown correlationID = int.from_bytes(buf[8:12], byteorder="big") diff --git a/src/stirling/source_connectors/socket_tracer/BUILD.bazel b/src/stirling/source_connectors/socket_tracer/BUILD.bazel index 76a9ca13b77..d6ab3aff991 100644 --- a/src/stirling/source_connectors/socket_tracer/BUILD.bazel +++ b/src/stirling/source_connectors/socket_tracer/BUILD.bazel @@ -426,7 +426,6 @@ pl_cc_bpf_test( "//src/common/testing/test_utils:cc_library", "//src/stirling/source_connectors/socket_tracer/testing:cc_library", "//src/stirling/source_connectors/socket_tracer/testing/container_images:kafka_container", - "//src/stirling/source_connectors/socket_tracer/testing/container_images:zookeeper_container", ], ) diff --git a/src/stirling/source_connectors/socket_tracer/bcc_bpf/protocol_inference.h b/src/stirling/source_connectors/socket_tracer/bcc_bpf/protocol_inference.h index e6f89610b98..6e982be159a 100644 --- a/src/stirling/source_connectors/socket_tracer/bcc_bpf/protocol_inference.h +++ b/src/stirling/source_connectors/socket_tracer/bcc_bpf/protocol_inference.h @@ -391,8 +391,8 @@ static __inline enum message_type_t infer_mysql_message(const char* buf, size_t // correlation_id => INT32 static __inline enum message_type_t infer_kafka_request(const char* buf) { // API is Kafka's terminology for opcode. - static const int kNumAPIs = 62; - static const int kMaxAPIVersion = 12; + static const int kNumAPIs = 93; + static const int kMaxAPIVersion = 18; const int16_t request_API_key = read_big_endian_int16(buf); if (request_API_key < 0 || request_API_key > kNumAPIs) { diff --git a/src/stirling/source_connectors/socket_tracer/kafka_trace_bpf_test.cc b/src/stirling/source_connectors/socket_tracer/kafka_trace_bpf_test.cc index 733c0c41f34..b36e09a42e9 100644 --- a/src/stirling/source_connectors/socket_tracer/kafka_trace_bpf_test.cc +++ b/src/stirling/source_connectors/socket_tracer/kafka_trace_bpf_test.cc @@ -21,6 +21,7 @@ #include #include +#include #include #include "src/common/base/base.h" @@ -29,7 +30,6 @@ #include "src/common/testing/testing.h" #include "src/stirling/source_connectors/socket_tracer/protocols/kafka/common/types.h" #include "src/stirling/source_connectors/socket_tracer/testing/container_images/kafka_container.h" -#include "src/stirling/source_connectors/socket_tracer/testing/container_images/zookeeper_container.h" #include "src/stirling/source_connectors/socket_tracer/testing/socket_trace_bpf_test_fixture.h" #include "src/stirling/testing/common.h" @@ -46,27 +46,39 @@ using ::testing::Contains; using ::testing::Eq; using ::testing::Field; using ::testing::HasSubstr; -using ::testing::StrEq; using ::px::operator<<; +// Modern Kafka brokers (Kafka 3.1+) negotiate protocol versions well above what Pixie originally +// supported (e.g. Fetch v13+ with topic-id UUIDs, and higher Produce/ApiVersions/Metadata +// versions). This test runs against both the Confluent (confluentinc/cp-kafka) and Apache +// (apache/kafka) broker distributions in KRaft mode to confirm that Produce/Fetch/ApiVersions +// traces are captured for these modern versions, which regressed prior to this change (see +// https://github.com/pixie-io/pixie/issues/2138). +template class KafkaTraceTest : public SocketTraceBPFTestFixture { protected: + static constexpr std::string_view kTopic = "foo"; + static constexpr std::string_view kBootstrapServer = "localhost:29092"; + KafkaTraceTest() { - // Run Zookeeper. - StatusOr zookeeper_run_result = zookeeper_server_.Run( - std::chrono::seconds{90}, - {"--name=zookeeper", "--env=ZOOKEEPER_CLIENT_PORT=32181", "--env=ZOOKEEPER_TICK_TIME=2000", - "--env=ZOOKEEPER_SYNC_LIMIT=2", "--env=ZOOKEEPER_ADMIN_SERVER_PORT=8020"}); - PX_CHECK_OK(zookeeper_run_result); - - // Run Kafka server. + // Run the Kafka broker in KRaft mode (no ZooKeeper). The env vars below are shared by both + // the Confluent and Apache images since both translate KAFKA_* env vars into server config. StatusOr kafka_run_result = kafka_server_.Run( - std::chrono::seconds{90}, - {absl::Substitute("--network=container:$0", zookeeper_server_.container_name()), - "--name=kafka", "--env=KAFKA_ZOOKEEPER_CONNECT=localhost:32181", + std::chrono::seconds{120}, + {"--env=CLUSTER_ID=MkU3OEVBNTcwNTJENDM2Qk", "--env=KAFKA_NODE_ID=1", + "--env=KAFKA_PROCESS_ROLES=broker,controller", + "--env=KAFKA_CONTROLLER_QUORUM_VOTERS=1@localhost:29093", + "--env=KAFKA_LISTENERS=PLAINTEXT://:29092,CONTROLLER://:29093", "--env=KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:29092", + "--env=KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER", + "--env=KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT", "--env=KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1"}); - PX_CHECK_OK(zookeeper_run_result); + PX_CHECK_OK(kafka_run_result); + } + + // Builds the path to a Kafka CLI tool for the broker distribution under test. + std::string Tool(std::string_view name) { + return absl::StrCat(TKafkaContainer::kBinPath, name, TKafkaContainer::kToolSuffix); } StatusOr GetPIDFromOutput(std::string_view out) { @@ -85,9 +97,9 @@ class KafkaTraceTest : public SocketTraceBPFTestFixture CreateTopic() { std::string cmd = absl::StrFormat( - "podman exec %s bash -c 'kafka-topics --create --topic foo --partitions 1 " - "--replication-factor 1 --if-not-exists --zookeeper localhost:32181 & echo $! && wait'", - kafka_server_.container_name()); + "podman exec %s bash -c '%s --create --topic %s --partitions 1 " + "--replication-factor 1 --if-not-exists --bootstrap-server %s & echo $! && wait'", + kafka_server_.container_name(), Tool("kafka-topics"), kTopic, kBootstrapServer); PX_ASSIGN_OR_RETURN(std::string out, px::Exec(cmd)); return GetPIDFromOutput(out); @@ -96,9 +108,9 @@ class KafkaTraceTest : public SocketTraceBPFTestFixture ProduceMessage() { std::string cmd = absl::StrFormat( "podman exec %s bash -c 'echo \"hello\" | " - "kafka-console-producer --request-required-acks 1 --broker-list localhost:29092 --topic " - "foo& echo $! && wait'", - kafka_server_.container_name()); + "%s --request-required-acks 1 --bootstrap-server %s --topic " + "%s& echo $! && wait'", + kafka_server_.container_name(), Tool("kafka-console-producer"), kBootstrapServer, kTopic); PX_ASSIGN_OR_RETURN(std::string out, px::Exec(cmd)); return GetPIDFromOutput(out); @@ -106,16 +118,15 @@ class KafkaTraceTest : public SocketTraceBPFTestFixture FetchMessage() { std::string cmd = absl::StrFormat( - "podman exec %s bash -c 'kafka-console-consumer --bootstrap-server localhost:29092 --topic " - "foo --from-beginning --timeout-ms 10000& echo $! && wait'", - kafka_server_.container_name()); + "podman exec %s bash -c '%s --bootstrap-server %s --topic " + "%s --from-beginning --timeout-ms 10000& echo $! && wait'", + kafka_server_.container_name(), Tool("kafka-console-consumer"), kBootstrapServer, kTopic); PX_ASSIGN_OR_RETURN(std::string out, px::Exec(cmd)); return GetPIDFromOutput(out); } - ::px::stirling::testing::KafkaContainer kafka_server_; - ::px::stirling::testing::ZooKeeperContainer zookeeper_server_; + TKafkaContainer kafka_server_; }; struct KafkaTraceRecord { @@ -131,105 +142,6 @@ struct KafkaTraceRecord { } }; -auto EqKafkaTraceRecord(const KafkaTraceRecord& x) { - return AllOf(Field(&KafkaTraceRecord::req_cmd, Eq(x.req_cmd)), - // client_id is dynamic for the consumer. - Field(&KafkaTraceRecord::client_id, HasSubstr(x.client_id)), - Field(&KafkaTraceRecord::req_body, StrEq(x.req_body)), - Field(&KafkaTraceRecord::resp, StrEq(x.resp))); -} - -KafkaTraceRecord kLeaderAndIsrRecord = { - .req_cmd = kafka::APIKey::kLeaderAndIsr, .client_id = "1001", .req_body = "", .resp = ""}; - -KafkaTraceRecord kUpdateMetadataRecord = { - .req_cmd = kafka::APIKey::kUpdateMetadata, .client_id = "1001", .req_body = "", .resp = ""}; - -KafkaTraceRecord kProducerMetadataRecord = {.req_cmd = kafka::APIKey::kMetadata, - .client_id = "console-producer", - .req_body = "", - .resp = ""}; - -KafkaTraceRecord kConsumerMetadataRecord = {.req_cmd = kafka::APIKey::kMetadata, - .client_id = "console-consumer", - .req_body = "", - .resp = ""}; - -KafkaTraceRecord kFindCoordinatorRecord = {.req_cmd = kafka::APIKey::kFindCoordinator, - .client_id = "console-consumer", - .req_body = "", - .resp = ""}; - -KafkaTraceRecord kProducerApiVersionsRecord = {.req_cmd = kafka::APIKey::kApiVersions, - .client_id = "console-producer", - .req_body = "", - .resp = ""}; - -KafkaTraceRecord kConsumerApiVersionsRecord = {.req_cmd = kafka::APIKey::kApiVersions, - .client_id = "console-consumer", - .req_body = "", - .resp = ""}; - -KafkaTraceRecord kJoinGroupRecord = { - .req_cmd = kafka::APIKey::kJoinGroup, - .client_id = "console-consumer", - .req_body = - "{\"group_id\":\"console-consumer\",\"session_timeout_ms\":10000,\"rebalance_timeout_ms\":" - "300000,\"member_id\":\"consumer-console-consumer\",\"group_instance_id\":\"\",\"protocol_" - "type\":\"consumer\",\"protocols\":[{\"protocol\":\"range\"}]}", - .resp = - "{\"throttle_time_ms\":0,\"error_code\":0,\"generation_id\":1,\"protocol_type\":" - "\"consumer\",\"protocol_name\":\"range\",\"leader\":\"consumer-console-consumer\"," - "\"member_id\":\"consumer-console-consumer\",\"members\":[{\"member_id\":\"consumer-" - "console-consumer\",\"group_instance_id\":\"\"}]}"}; - -KafkaTraceRecord kSyncGroupRecord = { - .req_cmd = kafka::APIKey::kSyncGroup, - .client_id = "console-consumer", - .req_body = - "{\"group_id\":\"console-consumer\",\"generation_id\":1,\"member_id\":\"consumer-console-" - "consumer\",\"group_instance_id\":\"\",\"protocol_type\":\"consumer\",\"protocol_name\":" - "\"range\",\"assignments\":[{\"member_id\":\"consumer-console-consumer\"}]}", - .resp = - "{\"throttle_time_ms\":0,\"error_code\":0,\"protocol_type\":\"consumer\",\"protocol_name\":" - "\"range\"}"}; - -KafkaTraceRecord kProduceRecord = { - .req_cmd = kafka::APIKey::kProduce, - .client_id = "console-producer", - .req_body = - "{\"transactional_id\":\"\",\"acks\":1,\"timeout_ms\":1500,\"topics\":[{\"name\":\"foo\"," - "\"partitions\":[{\"index\":0,\"message_set\":{\"size\":74}}]}]}", - .resp = - "{\"topics\":[{\"name\":\"foo\",\"partitions\":[{\"index\":0,\"error_code\":0," - "\"base_offset\":0,\"log_append_time_ms\":-1,\"log_start_offset\":0,\"record_errors\":[]," - "\"error_message\":\"\"}]}],\"throttle_time_ms\":0}"}; - -KafkaTraceRecord kOffsetFetchRecord = {.req_cmd = kafka::APIKey::kOffsetFetch, - .client_id = "console-consumer", - .req_body = "", - .resp = ""}; - -KafkaTraceRecord kListOffsetsRecord = {.req_cmd = kafka::APIKey::kListOffsets, - .client_id = "console-consumer", - .req_body = "", - .resp = ""}; - -KafkaTraceRecord kFetchRecord = { - .req_cmd = kafka::APIKey::kFetch, - .client_id = "console-consumer", - .req_body = - "{\"replica_id\":-1,\"session_id\":0,\"session_epoch\":0,\"topics\":[{\"name\":\"foo\"," - "\"partitions\":[{\"index\":0,\"current_leader_epoch\":0,\"fetch_offset\":0,\"last_fetched_" - "epoch\":-1,\"log_start_offset\":-1,\"partition_max_bytes\":1048576}]}],\"forgotten_" - "topics\":[],\"rack_id\":\"\"}", - .resp = - "{\"throttle_time_ms\":0,\"error_code\":0,\"session_id\":,\"topics\":[{\"name\":" - "\"foo\",\"partitions\":[{" - "\"index\":0,\"error_code\":0,\"high_" - "watermark\":1,\"last_stable_offset\":1,\"log_start_offset\":0,\"aborted_transactions\":[]," - "\"preferred_read_replica\":-1,\"message_set\":{\"size\":74}}]}]}"}; - std::vector GetKafkaTraceRecords( const types::ColumnWrapperRecordBatch& record_batch, int pid) { std::vector res; @@ -253,55 +165,77 @@ std::vector GetKafkaTraceRecords( return res; } +// Matches a record by request command only. +auto EqKafkaCmd(kafka::APIKey req_cmd) { + return Field(&KafkaTraceRecord::req_cmd, Eq(req_cmd)); +} + +// Matches a record by request command, with a substring expected in the request body. +auto EqKafkaCmdWithReqBody(kafka::APIKey req_cmd, std::string_view req_body_substr) { + return AllOf(Field(&KafkaTraceRecord::req_cmd, Eq(req_cmd)), + Field(&KafkaTraceRecord::req_body, HasSubstr(std::string(req_body_substr)))); +} + +// Matches a record by request command, with a substring expected in the response. +auto EqKafkaCmdWithResp(kafka::APIKey req_cmd, std::string_view resp_substr) { + return AllOf(Field(&KafkaTraceRecord::req_cmd, Eq(req_cmd)), + Field(&KafkaTraceRecord::resp, HasSubstr(std::string(resp_substr)))); +} + +using KafkaContainerTypes = ::testing::Types<::px::stirling::testing::KafkaContainer, + ::px::stirling::testing::ApacheKafkaContainer>; +TYPED_TEST_SUITE(KafkaTraceTest, KafkaContainerTypes); + //----------------------------------------------------------------------------- // Test Scenarios //----------------------------------------------------------------------------- -TEST_F(KafkaTraceTest, kafka_capture) { - StartTransferDataThread(); +TYPED_TEST(KafkaTraceTest, kafka_capture) { + this->StartTransferDataThread(); - ASSERT_OK_AND_ASSIGN(int32_t create_topic_pid, CreateTopic()); + ASSERT_OK_AND_ASSIGN(int32_t create_topic_pid, this->CreateTopic()); PX_UNUSED(create_topic_pid); - ASSERT_OK_AND_ASSIGN(int32_t produce_message_pid, ProduceMessage()); - ASSERT_OK_AND_ASSIGN(int32_t fetch_message_pid, FetchMessage()); + ASSERT_OK_AND_ASSIGN(int32_t produce_message_pid, this->ProduceMessage()); + ASSERT_OK_AND_ASSIGN(int32_t fetch_message_pid, this->FetchMessage()); - StopTransferDataThread(); + this->StopTransferDataThread(); // Grab the data from Stirling. - std::vector tablets = ConsumeRecords(SocketTraceConnector::kKafkaTableNum); + std::vector tablets = + this->ConsumeRecords(SocketTraceConnector::kKafkaTableNum); ASSERT_NOT_EMPTY_AND_GET_RECORDS(const types::ColumnWrapperRecordBatch& record_batch, tablets); - // TODO(chengruizhe): Some of the records are missing. Fix and add tests for all records. - // TODO(vsrivatsa): enable kafka Metadata test once Metadata response parsing implemeneted + + // Broker (server) side: it processes both the produce and fetch traffic. On modern brokers the + // fetch uses topic-id UUIDs (api_version >= 13), which previously failed to parse. { - auto records = GetKafkaTraceRecords(record_batch, kafka_server_.process_pid()); - // ApiVersion requests are dropped by the server, since they are the first packets. - EXPECT_THAT(records, Contains(EqKafkaTraceRecord(kLeaderAndIsrRecord))); - EXPECT_THAT(records, Contains(EqKafkaTraceRecord(kUpdateMetadataRecord))); - EXPECT_THAT(records, Contains(EqKafkaTraceRecord(kProduceRecord))); - // EXPECT_THAT(records, Contains(EqKafkaTraceRecord(kConsumerMetadataRecord))); - EXPECT_THAT(records, Contains(EqKafkaTraceRecord(kFindCoordinatorRecord))); - EXPECT_THAT(records, Contains(EqKafkaTraceRecord(kJoinGroupRecord))); - EXPECT_THAT(records, Contains(EqKafkaTraceRecord(kSyncGroupRecord))); - EXPECT_THAT(records, Contains(EqKafkaTraceRecord(kOffsetFetchRecord))); - EXPECT_THAT(records, Contains(EqKafkaTraceRecord(kListOffsetsRecord))); - EXPECT_THAT(records, Contains(EqKafkaTraceRecord(kFetchRecord))); + auto records = GetKafkaTraceRecords(record_batch, this->kafka_server_.process_pid()); + EXPECT_THAT(records, Contains(EqKafkaCmdWithReqBody(kafka::APIKey::kProduce, "\"name\":\"foo\""))) + << "Expected a Produce record referencing topic foo from the broker."; + EXPECT_THAT(records, Contains(EqKafkaCmdWithResp(kafka::APIKey::kFetch, "message_set"))) + << "Expected a Fetch record with a decoded message_set from the broker."; + EXPECT_THAT(records, Contains(EqKafkaCmd(kafka::APIKey::kFindCoordinator))); + EXPECT_THAT(records, Contains(EqKafkaCmd(kafka::APIKey::kListOffsets))); } + + // Producer client: sends ApiVersions (negotiation) then Produce. { auto records = GetKafkaTraceRecords(record_batch, produce_message_pid); - EXPECT_THAT(records, Contains(EqKafkaTraceRecord(kProducerApiVersionsRecord))); - // EXPECT_THAT(records, Contains(EqKafkaTraceRecord(kProducerMetadataRecord))); - EXPECT_THAT(records, Contains(EqKafkaTraceRecord(kProduceRecord))); + EXPECT_THAT(records, Contains(EqKafkaCmd(kafka::APIKey::kApiVersions))) + << "Expected an ApiVersions record from the producer."; + EXPECT_THAT(records, Contains(EqKafkaCmdWithReqBody(kafka::APIKey::kProduce, "\"name\":\"foo\""))) + << "Expected a Produce record referencing topic foo from the producer."; } + + // Consumer client: sends ApiVersions then Fetch. The fetch request/response carry a topic_id + // UUID (api_version >= 13). Capturing this record is the core regression fix. { auto records = GetKafkaTraceRecords(record_batch, fetch_message_pid); - EXPECT_THAT(records, Contains(EqKafkaTraceRecord(kConsumerApiVersionsRecord))); - // EXPECT_THAT(records, Contains(EqKafkaTraceRecord(kConsumerMetadataRecord))); - EXPECT_THAT(records, Contains(EqKafkaTraceRecord(kFindCoordinatorRecord))); - EXPECT_THAT(records, Contains(EqKafkaTraceRecord(kJoinGroupRecord))); - EXPECT_THAT(records, Contains(EqKafkaTraceRecord(kSyncGroupRecord))); - EXPECT_THAT(records, Contains(EqKafkaTraceRecord(kOffsetFetchRecord))); - EXPECT_THAT(records, Contains(EqKafkaTraceRecord(kListOffsetsRecord))); - EXPECT_THAT(records, Contains(EqKafkaTraceRecord(kFetchRecord))); + EXPECT_THAT(records, Contains(EqKafkaCmd(kafka::APIKey::kApiVersions))) + << "Expected an ApiVersions record from the consumer."; + EXPECT_THAT(records, Contains(EqKafkaCmdWithReqBody(kafka::APIKey::kFetch, "topic_id"))) + << "Expected a Fetch record with a decoded topic_id (UUID) from the consumer."; + EXPECT_THAT(records, Contains(EqKafkaCmdWithResp(kafka::APIKey::kFetch, "message_set"))) + << "Expected a Fetch response with a decoded message_set from the consumer."; } } diff --git a/src/stirling/source_connectors/socket_tracer/protocols/kafka/common/types.h b/src/stirling/source_connectors/socket_tracer/protocols/kafka/common/types.h index a7e1820d312..53bbf9ac133 100644 --- a/src/stirling/source_connectors/socket_tracer/protocols/kafka/common/types.h +++ b/src/stirling/source_connectors/socket_tracer/protocols/kafka/common/types.h @@ -97,10 +97,49 @@ enum class APIKey : int16_t { kAlterClientQuotas = 49, kDescribeUserScramCredentials = 50, kAlterUserScramCredentials = 51, + kVote = 52, + kBeginQuorumEpoch = 53, + kEndQuorumEpoch = 54, + kDescribeQuorum = 55, + // Renamed from AlterIsr to AlterPartition in Kafka 3.x, but the api key is unchanged. kAlterIsr = 56, kUpdateFeatures = 57, + kEnvelope = 58, + kFetchSnapshot = 59, kDescribeCluster = 60, kDescribeProducers = 61, + kBrokerRegistration = 62, + kBrokerHeartbeat = 63, + kUnregisterBroker = 64, + kDescribeTransactions = 65, + kListTransactions = 66, + kAllocateProducerIds = 67, + kConsumerGroupHeartbeat = 68, + kConsumerGroupDescribe = 69, + kControllerRegistration = 70, + kGetTelemetrySubscriptions = 71, + kPushTelemetry = 72, + kAssignReplicasToDirs = 73, + kListConfigResources = 74, + kDescribeTopicPartitions = 75, + kShareGroupHeartbeat = 76, + kShareGroupDescribe = 77, + kShareFetch = 78, + kShareAcknowledge = 79, + kAddRaftVoter = 80, + kRemoveRaftVoter = 81, + kUpdateRaftVoter = 82, + kInitializeShareGroupState = 83, + kReadShareGroupState = 84, + kWriteShareGroupState = 85, + kDeleteShareGroupState = 86, + kReadShareGroupStateSummary = 87, + kStreamsGroupHeartbeat = 88, + kStreamsGroupDescribe = 89, + kDescribeShareGroupOffsets = 90, + kAlterShareGroupOffsets = 91, + kDeleteShareGroupOffsets = 92, + kStreamsGroupTopologyDescriptionUpdate = 93, }; // Error Codes @@ -226,65 +265,106 @@ struct APIVersionData { // https://cwiki.apache.org/confluence/display/KAFKA/KIP-482%3A+The+Kafka+Protocol+should+Support+Optional+Tagged+Fields#KIP482:TheKafkaProtocolshouldSupportOptionalTaggedFields-FlexibleVersions // Detailed information on each API key: // https://github.com/apache/kafka/tree/trunk/clients/src/main/resources/common/message -// TODO(chengruizhe): Needs updating for new opcodes. +// Versions below are refreshed against Kafka trunk (4.4.0-SNAPSHOT), which is a superset of all +// released Kafka protocol versions through 4.x. Min versions are intentionally kept low (0, or 1 +// for Produce) to make frame boundary detection more robust; only the max and flexible versions +// need to track upstream to avoid rejecting modern clients. inline const absl::flat_hash_map APIVersionMap = { // Setting min supported version to 1 to help finding frame boundary. - {APIKey::kProduce, {1, 9, 9}}, - {APIKey::kFetch, {0, 12, 12}}, - {APIKey::kListOffsets, {0, 7, 6}}, - {APIKey::kMetadata, {0, 12, 9}}, + {APIKey::kProduce, {1, 13, 9}}, + {APIKey::kFetch, {0, 18, 12}}, + {APIKey::kListOffsets, {0, 11, 6}}, + {APIKey::kMetadata, {0, 13, 9}}, {APIKey::kLeaderAndIsr, {0, 5, 4}}, {APIKey::kStopReplica, {0, 3, 2}}, {APIKey::kUpdateMetadata, {0, 7, 6}}, {APIKey::kControlledShutdown, {0, 3, 3}}, - {APIKey::kOffsetCommit, {0, 8, 8}}, - {APIKey::kOffsetFetch, {0, 8, 6}}, - {APIKey::kFindCoordinator, {0, 4, 3}}, - {APIKey::kJoinGroup, {0, 7, 6}}, + {APIKey::kOffsetCommit, {0, 10, 8}}, + {APIKey::kOffsetFetch, {0, 10, 6}}, + {APIKey::kFindCoordinator, {0, 6, 3}}, + {APIKey::kJoinGroup, {0, 9, 6}}, {APIKey::kHeartbeat, {0, 4, 4}}, - {APIKey::kLeaveGroup, {0, 4, 4}}, + {APIKey::kLeaveGroup, {0, 5, 4}}, {APIKey::kSyncGroup, {0, 5, 4}}, - {APIKey::kDescribeGroups, {0, 5, 5}}, - {APIKey::kListGroups, {0, 4, 3}}, + {APIKey::kDescribeGroups, {0, 6, 5}}, + {APIKey::kListGroups, {0, 5, 3}}, {APIKey::kSaslHandshake, {0, 1, -1}}, - {APIKey::kApiVersions, {0, 3, 3}}, + {APIKey::kApiVersions, {0, 5, 3}}, {APIKey::kCreateTopics, {0, 7, 5}}, {APIKey::kDeleteTopics, {0, 6, 4}}, {APIKey::kDeleteRecords, {0, 2, 2}}, - {APIKey::kInitProducerId, {0, 4, 2}}, + {APIKey::kInitProducerId, {0, 6, 2}}, {APIKey::kOffsetForLeaderEpoch, {0, 4, 4}}, - {APIKey::kAddPartitionsToTxn, {0, 3, 3}}, - {APIKey::kAddOffsetsToTxn, {0, 3, 3}}, - {APIKey::kEndTxn, {0, 3, 3}}, - {APIKey::kWriteTxnMarkers, {0, 1, 1}}, - {APIKey::kTxnOffsetCommit, {0, 3, 3}}, - {APIKey::kDescribeAcls, {0, 2, 2}}, - {APIKey::kCreateAcls, {0, 2, 2}}, - {APIKey::kDeleteAcls, {0, 2, 2}}, + {APIKey::kAddPartitionsToTxn, {0, 5, 3}}, + {APIKey::kAddOffsetsToTxn, {0, 4, 3}}, + {APIKey::kEndTxn, {0, 5, 3}}, + {APIKey::kWriteTxnMarkers, {0, 2, 1}}, + {APIKey::kTxnOffsetCommit, {0, 6, 3}}, + {APIKey::kDescribeAcls, {0, 3, 2}}, + {APIKey::kCreateAcls, {0, 3, 2}}, + {APIKey::kDeleteAcls, {0, 3, 2}}, {APIKey::kDescribeConfigs, {0, 4, 4}}, {APIKey::kAlterConfigs, {0, 2, 2}}, {APIKey::kAlterReplicaLogDirs, {0, 2, 2}}, - {APIKey::kDescribeLogDirs, {0, 2, 2}}, + {APIKey::kDescribeLogDirs, {0, 5, 2}}, {APIKey::kSaslAuthenticate, {0, 2, 2}}, {APIKey::kCreatePartitions, {0, 3, 2}}, - {APIKey::kCreateDelegationToken, {0, 2, 2}}, + {APIKey::kCreateDelegationToken, {0, 3, 2}}, {APIKey::kRenewDelegationToken, {0, 2, 2}}, {APIKey::kExpireDelegationToken, {0, 2, 2}}, - {APIKey::kDescribeDelegationToken, {0, 2, 2}}, - {APIKey::kDeleteGroups, {0, 5, 5}}, + {APIKey::kDescribeDelegationToken, {0, 3, 2}}, + {APIKey::kDeleteGroups, {0, 3, 2}}, {APIKey::kElectLeaders, {0, 2, 2}}, {APIKey::kIncrementalAlterConfigs, {0, 1, 1}}, - {APIKey::kAlterPartitionReassignments, {0, 0, 0}}, + {APIKey::kAlterPartitionReassignments, {0, 1, 0}}, {APIKey::kListPartitionReassignments, {0, 0, 0}}, {APIKey::kOffsetDelete, {0, 0, -1}}, {APIKey::kDescribeClientQuotas, {0, 1, 1}}, {APIKey::kAlterClientQuotas, {0, 1, 1}}, {APIKey::kDescribeUserScramCredentials, {0, 0, 0}}, {APIKey::kAlterUserScramCredentials, {0, 0, 0}}, - {APIKey::kAlterIsr, {0, 0, 0}}, - {APIKey::kUpdateFeatures, {0, 0, 0}}, - {APIKey::kDescribeCluster, {0, 0, 0}}, - {APIKey::kDescribeProducers, {0, 0, 0}}}; + {APIKey::kVote, {0, 2, 0}}, + {APIKey::kBeginQuorumEpoch, {0, 1, 1}}, + {APIKey::kEndQuorumEpoch, {0, 1, 1}}, + {APIKey::kDescribeQuorum, {0, 2, 0}}, + {APIKey::kAlterIsr, {0, 3, 0}}, + {APIKey::kUpdateFeatures, {0, 2, 0}}, + {APIKey::kEnvelope, {0, 0, 0}}, + {APIKey::kFetchSnapshot, {0, 1, 0}}, + {APIKey::kDescribeCluster, {0, 2, 0}}, + {APIKey::kDescribeProducers, {0, 0, 0}}, + {APIKey::kBrokerRegistration, {0, 4, 0}}, + {APIKey::kBrokerHeartbeat, {0, 2, 0}}, + {APIKey::kUnregisterBroker, {0, 0, 0}}, + {APIKey::kDescribeTransactions, {0, 0, 0}}, + {APIKey::kListTransactions, {0, 2, 0}}, + {APIKey::kAllocateProducerIds, {0, 0, 0}}, + {APIKey::kConsumerGroupHeartbeat, {0, 1, 0}}, + {APIKey::kConsumerGroupDescribe, {0, 1, 0}}, + {APIKey::kControllerRegistration, {0, 0, 0}}, + {APIKey::kGetTelemetrySubscriptions, {0, 0, 0}}, + {APIKey::kPushTelemetry, {0, 0, 0}}, + {APIKey::kAssignReplicasToDirs, {0, 0, 0}}, + {APIKey::kListConfigResources, {0, 1, 0}}, + {APIKey::kDescribeTopicPartitions, {0, 0, 0}}, + {APIKey::kShareGroupHeartbeat, {0, 1, 0}}, + {APIKey::kShareGroupDescribe, {0, 1, 0}}, + {APIKey::kShareFetch, {0, 2, 0}}, + {APIKey::kShareAcknowledge, {0, 2, 0}}, + {APIKey::kAddRaftVoter, {0, 1, 0}}, + {APIKey::kRemoveRaftVoter, {0, 0, 0}}, + {APIKey::kUpdateRaftVoter, {0, 0, 0}}, + {APIKey::kInitializeShareGroupState, {0, 0, 0}}, + {APIKey::kReadShareGroupState, {0, 0, 0}}, + {APIKey::kWriteShareGroupState, {0, 1, 0}}, + {APIKey::kDeleteShareGroupState, {0, 0, 0}}, + {APIKey::kReadShareGroupStateSummary, {0, 1, 0}}, + {APIKey::kStreamsGroupHeartbeat, {0, 1, 0}}, + {APIKey::kStreamsGroupDescribe, {0, 1, 0}}, + {APIKey::kDescribeShareGroupOffsets, {0, 1, 0}}, + {APIKey::kAlterShareGroupOffsets, {0, 0, 0}}, + {APIKey::kDeleteShareGroupOffsets, {0, 0, 0}}, + {APIKey::kStreamsGroupTopologyDescriptionUpdate, {0, 0, 0}}}; inline bool IsFlexible(APIKey api_key, int16_t api_version) { auto it = APIVersionMap.find(api_key); @@ -324,7 +404,7 @@ constexpr int kMinReqPacketLength = kMessageLengthBytes + kAPIKeyLength + kAPIVersionLength + kCorrelationIDLength; // length, correlation_id constexpr int kMinRespPacketLength = kMessageLengthBytes + kCorrelationIDLength; -constexpr int kMaxAPIVersion = 12; +constexpr int kMaxAPIVersion = 18; struct Packet : public FrameBase { int32_t correlation_id; diff --git a/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/fetch.cc b/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/fetch.cc index ff7ef20fbe9..98e2a4eedb7 100644 --- a/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/fetch.cc +++ b/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/fetch.cc @@ -26,7 +26,12 @@ namespace kafka { StatusOr PacketDecoder::ExtractFetchReqTopic() { FetchReqTopic r; - PX_ASSIGN_OR_RETURN(r.name, ExtractString()); + // In api_version >= 13, the topic is identified by a UUID instead of a name (KIP-516). + if (api_version_ >= 13) { + PX_ASSIGN_OR_RETURN(r.topic_id, ExtractUUID()); + } else { + PX_ASSIGN_OR_RETURN(r.name, ExtractString()); + } PX_ASSIGN_OR_RETURN(r.partitions, ExtractArray(&PacketDecoder::ExtractFetchReqPartition)); PX_RETURN_IF_ERROR(/* tag_section */ ExtractTagSection()); @@ -53,7 +58,12 @@ StatusOr PacketDecoder::ExtractFetchReqPartition() { StatusOr PacketDecoder::ExtractFetchForgottenTopicsData() { FetchForgottenTopicsData r; - PX_ASSIGN_OR_RETURN(r.name, ExtractString()); + // In api_version >= 13, the topic is identified by a UUID instead of a name (KIP-516). + if (api_version_ >= 13) { + PX_ASSIGN_OR_RETURN(r.topic_id, ExtractUUID()); + } else { + PX_ASSIGN_OR_RETURN(r.name, ExtractString()); + } PX_ASSIGN_OR_RETURN(r.partition_indices, ExtractArray(&PacketDecoder::ExtractInt32)); PX_RETURN_IF_ERROR(/* tag_section */ ExtractTagSection()); return r; @@ -61,7 +71,11 @@ StatusOr PacketDecoder::ExtractFetchForgottenTopicsDat StatusOr PacketDecoder::ExtractFetchReq() { FetchReq r; - PX_ASSIGN_OR_RETURN(r.replica_id, ExtractInt32()); + // In api_version >= 15, the top-level ReplicaId field was replaced by a tagged ReplicaState + // field (KIP-903), so it is no longer read here; it is skipped as part of the tag section. + if (api_version_ <= 14) { + PX_ASSIGN_OR_RETURN(r.replica_id, ExtractInt32()); + } PX_RETURN_IF_ERROR(/*max_wait_ms*/ ExtractInt32()); PX_RETURN_IF_ERROR(/*min_bytes*/ ExtractInt32()); @@ -127,7 +141,12 @@ StatusOr PacketDecoder::ExtractFetchRespPartition() { StatusOr PacketDecoder::ExtractFetchRespTopic() { FetchRespTopic r; - PX_ASSIGN_OR_RETURN(r.name, ExtractString()); + // In api_version >= 13, the topic is identified by a UUID instead of a name (KIP-516). + if (api_version_ >= 13) { + PX_ASSIGN_OR_RETURN(r.topic_id, ExtractUUID()); + } else { + PX_ASSIGN_OR_RETURN(r.name, ExtractString()); + } PX_ASSIGN_OR_RETURN(r.partitions, ExtractArray(&PacketDecoder::ExtractFetchRespPartition)); PX_RETURN_IF_ERROR(/* tag_section */ ExtractTagSection()); return r; diff --git a/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/fetch_test.cc b/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/fetch_test.cc index 9b6d1b6551b..80f257c355d 100644 --- a/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/fetch_test.cc +++ b/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/fetch_test.cc @@ -47,6 +47,9 @@ bool operator==(const FetchReqTopic& lhs, const FetchReqTopic& rhs) { if (lhs.name != rhs.name) { return false; } + if (lhs.topic_id != rhs.topic_id) { + return false; + } if (lhs.partitions.size() != rhs.partitions.size()) { return false; } @@ -64,6 +67,9 @@ bool operator==(const FetchForgottenTopicsData& lhs, const FetchForgottenTopicsD if (lhs.name != rhs.name) { return false; } + if (lhs.topic_id != rhs.topic_id) { + return false; + } if (lhs.partition_indices.size() != rhs.partition_indices.size()) { return false; } @@ -143,6 +149,9 @@ bool operator==(const FetchRespTopic& lhs, const FetchRespTopic& rhs) { if (lhs.name != rhs.name) { return false; } + if (lhs.topic_id != rhs.topic_id) { + return false; + } if (lhs.partitions.size() != rhs.partitions.size()) { return false; } @@ -262,6 +271,74 @@ TEST(KafkaPacketDecoder, TestExtractFetchReqV12) { EXPECT_OK_AND_EQ(decoder.ExtractFetchReq(), expected_result); } +// In api_version >= 13, the topic is identified by a 16-byte UUID (topic_id) instead of a name. +// This input is TestExtractFetchReqV12 with the topic name replaced by a topic_id UUID. +TEST(KafkaPacketDecoder, TestExtractFetchReqV13) { + const std::string_view input = CreateStringView( + "\xff\xff\xff\xff\x00\x00\x01\xf4\x00\x00\x00\x01\x03\x20\x00\x00\x00\x00\x00\x00\x00\x00\x00" + "\x00\x00\x02\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f\x02\x00" + "\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xff\xff\xff\xff\xff\xff\xff\xff" + "\xff\xff\xff\xff\x00\x10\x00\x00\x00\x00\x01\x01\x00"); + + FetchReqPartition partition{ + .index = 0, + .current_leader_epoch = 0, + .fetch_offset = 0, + .log_start_offset = -1, + .partition_max_bytes = 1048576, + }; + FetchReqTopic topic{ + .topic_id = "00010203-0405-0607-0809-0a0b0c0d0e0f", + .partitions = {partition}, + }; + FetchReq expected_result{ + .replica_id = -1, + .session_id = 0, + .session_epoch = 0, + .topics = {topic}, + .forgotten_topics = {}, + .rack_id = "", + }; + PacketDecoder decoder(input); + decoder.SetAPIInfo(APIKey::kFetch, 13); + EXPECT_OK_AND_EQ(decoder.ExtractFetchReq(), expected_result); +} + +// In api_version >= 15, the top-level ReplicaId field was removed (moved to a tagged +// ReplicaState field), so the body no longer begins with it. This input is TestExtractFetchReqV13 +// with the leading 4-byte replica_id removed. +TEST(KafkaPacketDecoder, TestExtractFetchReqV15) { + const std::string_view input = CreateStringView( + "\x00\x00\x01\xf4\x00\x00\x00\x01\x03\x20\x00\x00\x00\x00\x00\x00\x00\x00\x00" + "\x00\x00\x02\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f\x02\x00" + "\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xff\xff\xff\xff\xff\xff\xff\xff" + "\xff\xff\xff\xff\x00\x10\x00\x00\x00\x00\x01\x01\x00"); + + FetchReqPartition partition{ + .index = 0, + .current_leader_epoch = 0, + .fetch_offset = 0, + .log_start_offset = -1, + .partition_max_bytes = 1048576, + }; + FetchReqTopic topic{ + .topic_id = "00010203-0405-0607-0809-0a0b0c0d0e0f", + .partitions = {partition}, + }; + FetchReq expected_result{ + // replica_id is not present in the wire format for v15+, so it keeps its default value. + .replica_id = 0, + .session_id = 0, + .session_epoch = 0, + .topics = {topic}, + .forgotten_topics = {}, + .rack_id = "", + }; + PacketDecoder decoder(input); + decoder.SetAPIInfo(APIKey::kFetch, 15); + EXPECT_OK_AND_EQ(decoder.ExtractFetchReq(), expected_result); +} + TEST(KafkaPacketDecoder, TestExtractFetchRespV4) { const std::string_view input = CreateStringView( "\x00\x00\x00\x00\x00\x00\x00\x01\x00\x08\x6D\x79\x2D\x74\x6F\x70\x69\x63\x00\x00\x00\x01" diff --git a/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/join_group.cc b/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/join_group.cc index c86f651c182..9ec0856fb2b 100644 --- a/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/join_group.cc +++ b/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/join_group.cc @@ -59,6 +59,12 @@ StatusOr PacketDecoder::ExtractJoinGroupReq() { PX_ASSIGN_OR_RETURN(r.protocol_type, ExtractString()); PX_ASSIGN_OR_RETURN(r.protocols, ExtractArray(&PacketDecoder::ExtractJoinGroupProtocol)); + // Reason was added in api_version >= 8 (nullable string). Consume it so any trailing + // tag section stays aligned; it is currently not surfaced in the trace. + if (api_version_ >= 8) { + PX_RETURN_IF_ERROR(/* reason */ ExtractNullableString()); + } + return r; } @@ -76,6 +82,11 @@ StatusOr PacketDecoder::ExtractJoinGroupResp() { PX_ASSIGN_OR_RETURN(r.protocol_type, ExtractString()); PX_ASSIGN_OR_RETURN(r.protocol_name, ExtractString()); } + // SkipAssignment was added in api_version >= 9 (bool). Consume it so the trailing fields + // stay aligned; it is currently not surfaced in the trace. + if (api_version_ >= 9) { + PX_RETURN_IF_ERROR(/* skip_assignment */ ExtractBool()); + } PX_ASSIGN_OR_RETURN(r.leader, ExtractString()); PX_ASSIGN_OR_RETURN(r.member_id, ExtractString()); PX_ASSIGN_OR_RETURN(r.members, ExtractArray(&PacketDecoder::ExtractJoinGroupMember)); diff --git a/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/metadata.cc b/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/metadata.cc index 312d30151b8..147b737705a 100644 --- a/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/metadata.cc +++ b/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/metadata.cc @@ -27,8 +27,10 @@ namespace kafka { StatusOr PacketDecoder::ExtractMetadataReqTopic() { MetadataReqTopic r; + // The topic id is a UUID (16 raw bytes), present in api_version >= 10 (KIP-516). Note that in + // Metadata the UUID is additive: the topic name is still present (nullable in v10+). if (api_version_ >= 10) { - PX_ASSIGN_OR_RETURN(r.topic_id, ExtractString()); + PX_ASSIGN_OR_RETURN(r.topic_id, ExtractUUID()); } if (api_version_ <= 9) { diff --git a/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/packet_decoder.cc b/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/packet_decoder.cc index aaa2e48869a..d0481ef711d 100644 --- a/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/packet_decoder.cc +++ b/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/packet_decoder.cc @@ -138,6 +138,26 @@ StatusOr PacketDecoder::ExtractNullableString() { return ExtractRegularNullableString(); } +StatusOr PacketDecoder::ExtractUUID() { + // A UUID is always 16 raw bytes, regardless of the flexible version encoding. + constexpr int kUUIDNumBytes = 16; + PX_ASSIGN_OR_RETURN(std::string raw, ExtractBytesCore(kUUIDNumBytes)); + + // Format as a canonical 8-4-4-4-12 lowercase hex string. + constexpr char kHexDigits[] = "0123456789abcdef"; + std::string out; + out.reserve(36); + for (int i = 0; i < kUUIDNumBytes; ++i) { + if (i == 4 || i == 6 || i == 8 || i == 10) { + out.push_back('-'); + } + uint8_t byte = static_cast(raw[i]); + out.push_back(kHexDigits[byte >> 4]); + out.push_back(kHexDigits[byte & 0x0f]); + } + return out; +} + StatusOr PacketDecoder::ExtractRegularBytes() { PX_ASSIGN_OR_RETURN(int32_t len, ExtractInt16()); return ExtractBytesCore(len); diff --git a/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/packet_decoder.h b/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/packet_decoder.h index 5d7ee487356..005af2dd4ff 100644 --- a/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/packet_decoder.h +++ b/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/packet_decoder.h @@ -95,6 +95,12 @@ class PacketDecoder { StatusOr ExtractString(); StatusOr ExtractNullableString(); + // Represents a type 4 immutable universally unique identifier (UUID). Encoded as 16 raw bytes, + // independent of the flexible version encoding. Returned as a canonical hex string + // (8-4-4-4-12). Introduced by KIP-516 (topic IDs) and used in Produce v13+, Fetch v13+, and + // Metadata v10+. + StatusOr ExtractUUID(); + StatusOr ExtractBytes(); StatusOr ExtractNullableBytes(); diff --git a/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/packet_decoder_test.cc b/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/packet_decoder_test.cc index 577e147b5c2..524159d319b 100644 --- a/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/packet_decoder_test.cc +++ b/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/packet_decoder_test.cc @@ -124,6 +124,31 @@ INSTANTIATE_TEST_SUITE_P( PacketDecoderTestCase{std::string("\xFF\xFF\xFF\xFF\xFF\xFF\xFF\xFF\xFF\x01", 10), LONG_MIN})); +TEST(KafkaPacketDecoderTest, ExtractUUID) { + { + // 16 raw bytes are read regardless of flexible version and formatted as a canonical UUID. + const std::string_view msg = CreateStringView( + "\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f"); + PacketDecoder decoder(msg); + EXPECT_OK_AND_EQ(decoder.ExtractUUID(), "00010203-0405-0607-0809-0a0b0c0d0e0f"); + } + + // The all-zero UUID (used to represent a null/unset topic id). + { + const std::string_view msg = CreateStringView( + "\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"); + PacketDecoder decoder(msg); + EXPECT_OK_AND_EQ(decoder.ExtractUUID(), "00000000-0000-0000-0000-000000000000"); + } + + // Not enough bytes for a full UUID. + { + const std::string_view msg = CreateStringView("\x00\x01\x02\x03"); + PacketDecoder decoder(msg); + EXPECT_FALSE(decoder.ExtractUUID().ok()); + } +} + TEST(KafkaPacketDecoderTest, ExtractCompactString) { { const std::string_view msg = CreateStringView( diff --git a/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/produce.cc b/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/produce.cc index 4c5581aa0c4..8d97fcae03a 100644 --- a/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/produce.cc +++ b/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/produce.cc @@ -34,7 +34,12 @@ StatusOr PacketDecoder::ExtractProduceReqPartition() { StatusOr PacketDecoder::ExtractProduceReqTopic() { ProduceReqTopic r; - PX_ASSIGN_OR_RETURN(r.name, ExtractString()); + // In api_version >= 13, the topic is identified by a UUID instead of a name (KIP-516). + if (api_version_ >= 13) { + PX_ASSIGN_OR_RETURN(r.topic_id, ExtractUUID()); + } else { + PX_ASSIGN_OR_RETURN(r.name, ExtractString()); + } PX_ASSIGN_OR_RETURN(r.partitions, ExtractArray(&PacketDecoder::ExtractProduceReqPartition)); PX_RETURN_IF_ERROR(/* tag_section */ ExtractTagSection()); return r; @@ -73,15 +78,14 @@ StatusOr PacketDecoder::ExtractProduceRespPartition() { StatusOr PacketDecoder::ExtractProduceRespTopic() { ProduceRespTopic r; - if (is_flexible_) { - PX_ASSIGN_OR_RETURN(r.name, ExtractCompactString()); - PX_ASSIGN_OR_RETURN(r.partitions, - ExtractCompactArray(&PacketDecoder::ExtractProduceRespPartition)); - PX_RETURN_IF_ERROR(/* tag_section */ ExtractTagSection()); + // In api_version >= 13, the topic is identified by a UUID instead of a name (KIP-516). + if (api_version_ >= 13) { + PX_ASSIGN_OR_RETURN(r.topic_id, ExtractUUID()); } else { PX_ASSIGN_OR_RETURN(r.name, ExtractString()); - PX_ASSIGN_OR_RETURN(r.partitions, ExtractArray(&PacketDecoder::ExtractProduceRespPartition)); } + PX_ASSIGN_OR_RETURN(r.partitions, ExtractArray(&PacketDecoder::ExtractProduceRespPartition)); + PX_RETURN_IF_ERROR(/* tag_section */ ExtractTagSection()); return r; } diff --git a/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/produce_test.cc b/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/produce_test.cc index 89b7aa50470..8549cf6eae3 100644 --- a/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/produce_test.cc +++ b/src/stirling/source_connectors/socket_tracer/protocols/kafka/decoder/produce_test.cc @@ -44,6 +44,9 @@ bool operator==(const ProduceReqTopic& lhs, const ProduceReqTopic& rhs) { if (lhs.name != rhs.name) { return false; } + if (lhs.topic_id != rhs.topic_id) { + return false; + } if (lhs.partitions.size() != rhs.partitions.size()) { return false; } @@ -123,6 +126,9 @@ bool operator==(const ProduceRespTopic& lhs, const ProduceRespTopic& rhs) { if (lhs.name != rhs.name) { return false; } + if (lhs.topic_id != rhs.topic_id) { + return false; + } if (lhs.partitions.size() != rhs.partitions.size()) { return false; } @@ -267,6 +273,28 @@ TEST(KafkaPacketDecoderTest, ExtractProduceRespV9) { EXPECT_OK_AND_EQ(decoder.ExtractProduceResp(), expected_result); } +// In api_version >= 13, the topic is identified by a 16-byte UUID (topic_id) instead of a name. +// This input is ExtractProduceRespV9 with the topic name replaced by a topic_id UUID. +TEST(KafkaPacketDecoderTest, ExtractProduceRespV13) { + const std::string_view input = CreateStringView( + "\x02\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f\x02\x00" + "\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xff\xff\xff\xff\xff\xff\xff\xff\x00" + "\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00"); + ProduceRespPartition partition{.index = 0, + .error_code = 0, + .base_offset = 0, + .log_append_time_ms = -1, + .log_start_offset = 0, + .record_errors = {}, + .error_message = ""}; + ProduceRespTopic topic{.topic_id = "00010203-0405-0607-0809-0a0b0c0d0e0f", + .partitions = {partition}}; + ProduceResp expected_result{.topics = {topic}, .throttle_time_ms = 0}; + PacketDecoder decoder(input); + decoder.SetAPIInfo(APIKey::kProduce, 13); + EXPECT_OK_AND_EQ(decoder.ExtractProduceResp(), expected_result); +} + } // namespace kafka } // namespace protocols } // namespace stirling diff --git a/src/stirling/source_connectors/socket_tracer/protocols/kafka/opcodes/fetch.h b/src/stirling/source_connectors/socket_tracer/protocols/kafka/opcodes/fetch.h index 6eac33f99d3..106de3dfb45 100644 --- a/src/stirling/source_connectors/socket_tracer/protocols/kafka/opcodes/fetch.h +++ b/src/stirling/source_connectors/socket_tracer/protocols/kafka/opcodes/fetch.h @@ -51,9 +51,14 @@ struct FetchReqPartition { struct FetchReqTopic { std::string name; + // Topic id (UUID) replaces the topic name in api_version >= 13 (KIP-516). + std::string topic_id; std::vector partitions; void ToJSON(utils::JSONObjectBuilder* builder) const { + if (!topic_id.empty()) { + builder->WriteKV("topic_id", topic_id); + } builder->WriteKV("name", name); builder->WriteKVArrayRecursive("partitions", partitions); } @@ -61,9 +66,14 @@ struct FetchReqTopic { struct FetchForgottenTopicsData { std::string name; + // Topic id (UUID) replaces the topic name in api_version >= 13 (KIP-516). + std::string topic_id; std::vector partition_indices; void ToJSON(utils::JSONObjectBuilder* builder) const { + if (!topic_id.empty()) { + builder->WriteKV("topic_id", topic_id); + } builder->WriteKV("name", name); builder->WriteKV("partitions_indices", partition_indices); } @@ -123,9 +133,14 @@ struct FetchRespPartition { struct FetchRespTopic { std::string name; + // Topic id (UUID) replaces the topic name in api_version >= 13 (KIP-516). + std::string topic_id; std::vector partitions; void ToJSON(utils::JSONObjectBuilder* builder) const { + if (!topic_id.empty()) { + builder->WriteKV("topic_id", topic_id); + } builder->WriteKV("name", name); builder->WriteKVArrayRecursive("partitions", partitions); } diff --git a/src/stirling/source_connectors/socket_tracer/protocols/kafka/opcodes/produce.h b/src/stirling/source_connectors/socket_tracer/protocols/kafka/opcodes/produce.h index 8701beb44e5..9d5ce5a0e26 100644 --- a/src/stirling/source_connectors/socket_tracer/protocols/kafka/opcodes/produce.h +++ b/src/stirling/source_connectors/socket_tracer/protocols/kafka/opcodes/produce.h @@ -45,9 +45,14 @@ struct ProduceReqPartition { struct ProduceReqTopic { std::string name; + // Topic id (UUID) replaces the topic name in api_version >= 13 (KIP-516). + std::string topic_id; std::vector partitions; void ToJSON(utils::JSONObjectBuilder* builder) const { + if (!topic_id.empty()) { + builder->WriteKV("topic_id", topic_id); + } builder->WriteKV("name", name); builder->WriteKVArrayRecursive("partitions", partitions); } @@ -100,9 +105,14 @@ struct ProduceRespPartition { struct ProduceRespTopic { std::string name; + // Topic id (UUID) replaces the topic name in api_version >= 13 (KIP-516). + std::string topic_id; std::vector partitions; void ToJSON(utils::JSONObjectBuilder* builder) const { + if (!topic_id.empty()) { + builder->WriteKV("topic_id", topic_id); + } builder->WriteKV("name", name); builder->WriteKVArrayRecursive("partitions", partitions); } diff --git a/src/stirling/source_connectors/socket_tracer/testing/container_images/BUILD.bazel b/src/stirling/source_connectors/socket_tracer/testing/container_images/BUILD.bazel index 0d1ad990576..69cd6a3c1a2 100644 --- a/src/stirling/source_connectors/socket_tracer/testing/container_images/BUILD.bazel +++ b/src/stirling/source_connectors/socket_tracer/testing/container_images/BUILD.bazel @@ -101,6 +101,7 @@ pl_cc_test_library( srcs = [], hdrs = ["kafka_container.h"], data = [ + "//src/stirling/source_connectors/socket_tracer/testing/containers:apache_kafka_image.tar", "//src/stirling/source_connectors/socket_tracer/testing/containers:kafka_image.tar", ], deps = ["//src/common/testing/test_utils:cc_library"], @@ -355,12 +356,3 @@ pl_cc_test_library( deps = ["//src/common/testing/test_utils:cc_library"], ) -pl_cc_test_library( - name = "zookeeper_container", - srcs = [], - hdrs = ["zookeeper_container.h"], - data = [ - "//src/stirling/source_connectors/socket_tracer/testing/containers:zookeeper_image.tar", - ], - deps = ["//src/common/testing/test_utils:cc_library"], -) diff --git a/src/stirling/source_connectors/socket_tracer/testing/container_images/kafka_container.h b/src/stirling/source_connectors/socket_tracer/testing/container_images/kafka_container.h index ebb0751d686..ff6da84cfe9 100644 --- a/src/stirling/source_connectors/socket_tracer/testing/container_images/kafka_container.h +++ b/src/stirling/source_connectors/socket_tracer/testing/container_images/kafka_container.h @@ -27,18 +27,42 @@ namespace px { namespace stirling { namespace testing { +// Kafka broker packaged by Confluent (confluentinc/cp-kafka). Runs in KRaft mode +// (no ZooKeeper). The CLI tools (kafka-topics, kafka-console-producer, ...) are on the PATH. class KafkaContainer : public ContainerRunner { public: KafkaContainer() : ContainerRunner(::px::testing::BazelRunfilePath(kBazelImageTar), kContainerNamePrefix, kReadyMessage) {} + // Directory containing the CLI tools. Empty means they are on the PATH. + static constexpr std::string_view kBinPath = ""; + // Suffix on the CLI tool names (e.g. kafka-topics vs kafka-topics.sh). + static constexpr std::string_view kToolSuffix = ""; + private: static constexpr std::string_view kBazelImageTar = "src/stirling/source_connectors/socket_tracer/testing/containers/kafka_image.tar"; static constexpr std::string_view kContainerNamePrefix = "kafka_server"; - static constexpr std::string_view kReadyMessage = - "Recorded new controller, from now on will use broker"; + static constexpr std::string_view kReadyMessage = "Kafka Server started"; +}; + +// Kafka broker packaged by the Apache Kafka project (apache/kafka). Runs in KRaft mode. +// The CLI tools live under /opt/kafka/bin and carry a .sh suffix. +class ApacheKafkaContainer : public ContainerRunner { + public: + ApacheKafkaContainer() + : ContainerRunner(::px::testing::BazelRunfilePath(kBazelImageTar), kContainerNamePrefix, + kReadyMessage) {} + + static constexpr std::string_view kBinPath = "/opt/kafka/bin/"; + static constexpr std::string_view kToolSuffix = ".sh"; + + private: + static constexpr std::string_view kBazelImageTar = + "src/stirling/source_connectors/socket_tracer/testing/containers/apache_kafka_image.tar"; + static constexpr std::string_view kContainerNamePrefix = "apache_kafka_server"; + static constexpr std::string_view kReadyMessage = "Kafka Server started"; }; } // namespace testing diff --git a/src/stirling/source_connectors/socket_tracer/testing/container_images/zookeeper_container.h b/src/stirling/source_connectors/socket_tracer/testing/container_images/zookeeper_container.h deleted file mode 100644 index d0f59d48411..00000000000 --- a/src/stirling/source_connectors/socket_tracer/testing/container_images/zookeeper_container.h +++ /dev/null @@ -1,45 +0,0 @@ -/* - * Copyright 2018- The Pixie Authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - * SPDX-License-Identifier: Apache-2.0 - */ - -#pragma once - -#include - -#include "src/common/testing/test_environment.h" -#include "src/common/testing/test_utils/container_runner.h" - -namespace px { -namespace stirling { -namespace testing { - -class ZooKeeperContainer : public ContainerRunner { - public: - ZooKeeperContainer() - : ContainerRunner(::px::testing::BazelRunfilePath(kBazelImageTar), kContainerNamePrefix, - kReadyMessage) {} - - private: - static constexpr std::string_view kBazelImageTar = - "src/stirling/source_connectors/socket_tracer/testing/containers/zookeeper_image.tar"; - static constexpr std::string_view kContainerNamePrefix = "zookeeper_server"; - static constexpr std::string_view kReadyMessage = "INFO PrepRequestProcessor (sid:0) started"; -}; - -} // namespace testing -} // namespace stirling -} // namespace px diff --git a/src/stirling/source_connectors/socket_tracer/testing/containers/BUILD.bazel b/src/stirling/source_connectors/socket_tracer/testing/containers/BUILD.bazel index 3f73a0d5174..935a5480a60 100644 --- a/src/stirling/source_connectors/socket_tracer/testing/containers/BUILD.bazel +++ b/src/stirling/source_connectors/socket_tracer/testing/containers/BUILD.bazel @@ -108,8 +108,8 @@ container_image( ) container_image( - name = "zookeeper_image", - base = "@zookeeper_base_image//image", + name = "apache_kafka_image", + base = "@apache_kafka_base_image//image", ) [