diff --git a/source/common/common/logger.h b/source/common/common/logger.h
index 467aea5c2da42..35d2e2b74fe7e 100644
--- a/source/common/common/logger.h
+++ b/source/common/common/logger.h
@@ -34,6 +34,7 @@ namespace Logger {
   FUNCTION(http)                 \
   FUNCTION(http2)                \
   FUNCTION(hystrix)              \
+  FUNCTION(kafka)                \
   FUNCTION(lua)                  \
   FUNCTION(main)                 \
   FUNCTION(misc)                 \
diff --git a/source/extensions/extensions_build_config.bzl b/source/extensions/extensions_build_config.bzl
index 43cf6ea67f0fb..e69e813ac9c5a 100644
--- a/source/extensions/extensions_build_config.bzl
+++ b/source/extensions/extensions_build_config.bzl
@@ -65,6 +65,7 @@ EXTENSIONS = {
     "envoy.filters.network.echo":                       "//source/extensions/filters/network/echo:config",
     "envoy.filters.network.ext_authz":                  "//source/extensions/filters/network/ext_authz:config",
     "envoy.filters.network.http_connection_manager":    "//source/extensions/filters/network/http_connection_manager:config",
+    "envoy.filters.network.kafka":                      "//source/extensions/filters/network/kafka:config",
     "envoy.filters.network.mongo_proxy":                "//source/extensions/filters/network/mongo_proxy:config",
     "envoy.filters.network.ratelimit":                  "//source/extensions/filters/network/ratelimit:config",
     "envoy.filters.network.rbac":                       "//source/extensions/filters/network/rbac:config",
diff --git a/source/extensions/filters/network/kafka/BUILD b/source/extensions/filters/network/kafka/BUILD
new file mode 100644
index 0000000000000..52371c0d1f35b
--- /dev/null
+++ b/source/extensions/filters/network/kafka/BUILD
@@ -0,0 +1,118 @@
+licenses(["notice"])  # Apache 2
+
+# Kafka network filter.
+# Public docs: docs/root/configuration/network_filters/kafka_filter.rst
+
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_library",
+    "envoy_package",
+)
+
+envoy_package()
+
+envoy_cc_library(
+    name = "config",
+    srcs = ["config.cc"],
+    hdrs = ["metrics_holder.h"],
+    deps = [
+        ":kafka_filter_lib",
+        ":metrics_holder_lib",
+        "//include/envoy/registry",
+        "//include/envoy/server:filter_config_interface",
+        "//source/extensions/filters/network:well_known_names",
+    ],
+)
+
+envoy_cc_library(
+    name = "kafka_filter_lib",
+    srcs = ["kafka_filter.cc"],
+    hdrs = ["kafka_filter.h"],
+    deps = [
+        ":kafka_codec_lib",
+        ":kafka_request_lib",
+        ":kafka_response_lib",
+        ":metrics_holder_lib",
+        "//include/envoy/buffer:buffer_interface",
+        "//include/envoy/network:connection_interface",
+        "//include/envoy/network:filter_interface",
+        "//source/common/common:assert_lib",
+        "//source/common/common:minimal_logger_lib",
+    ],
+)
+
+envoy_cc_library(
+    name = "kafka_codec_lib",
+    srcs = ["codec.cc"],
+    hdrs = ["codec.h"],
+    deps = [
+        ":kafka_request_lib",
+        ":kafka_response_lib",
+        "//include/envoy/buffer:buffer_interface",
+    ],
+)
+
+envoy_cc_library(
+    name = "kafka_request_lib",
+    srcs = ["kafka_request.cc"],
+    hdrs = ["kafka_request.h"],
+    deps = [
+        ":parser_lib",
+        ":serialization_lib",
+        "//source/common/common:assert_lib",
+        "//source/common/common:minimal_logger_lib",
+    ],
+)
+
+envoy_cc_library(
+    name = "kafka_response_lib",
+    srcs = ["kafka_response.cc"],
+    hdrs = ["kafka_response.h"],
+    deps = [
+        ":parser_lib",
+        ":serialization_lib",
+        "//source/common/common:minimal_logger_lib",
+    ],
+)
+
+envoy_cc_library(
+    name = "metrics_holder_lib",
+    srcs = ["metrics_holder.cc"],
+    hdrs = ["metrics_holder.h"],
+    deps = [
+        ":kafka_protocol_lib",
+        "//include/envoy/stats:stats_interface",
+        "//source/common/common:macros",
+        "//source/common/common:to_lower_table_lib",
+    ],
+)
+
+envoy_cc_library(
+    name = "parser_lib",
+    hdrs = ["parser.h"],
+    deps = [
+        ":kafka_protocol_lib",
+        "//source/common/common:minimal_logger_lib",
+    ],
+)
+
+envoy_cc_library(
+    name = "serialization_lib",
+    hdrs = ["serialization.h"],
+    deps = [
+        ":kafka_protocol_lib",
+        "//source/common/common:byte_order_lib",
+    ],
+)
+
+envoy_cc_library(
+    name = "kafka_protocol_lib",
+    hdrs = [
+        "kafka_protocol.h",
+        "kafka_types.h",
+    ],
+    external_deps = ["abseil_optional"],
+    deps = [
+        "//source/common/common:macros",
+    ],
+)
diff --git a/source/extensions/filters/network/kafka/codec.cc b/source/extensions/filters/network/kafka/codec.cc
new file mode 100644
index 0000000000000..e380db14d8f58
--- /dev/null
+++ b/source/extensions/filters/network/kafka/codec.cc
@@ -0,0 +1,51 @@
+#include "extensions/filters/network/kafka/codec.h"
+
+#include "extensions/filters/network/kafka/kafka_protocol.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+
+void RequestDecoder::onData(Buffer::Instance& data) {
+  uint64_t num_slices = data.getRawSlices(nullptr, 0);
+  Buffer::RawSlice slices[num_slices];
+  data.getRawSlices(slices, num_slices);
+  for (const Buffer::RawSlice& slice : slices) {
+    doParse(request_parser_, slice);
+  }
+}
+
+void RequestDecoder::doParse(ParserSharedPtr& parser, const Buffer::RawSlice& slice) {
+  const char* buffer = reinterpret_cast<const char*>(slice.mem_);
+  uint64_t remaining = slice.len_;
+  while (remaining) {
+    ParseResponse result = parser->parse(buffer, remaining);
+    // this loop guarantees that parsers consuming 0 bytes also get processed
+    while (result.hasData()) {
+      if (!result.next_parser_) {
+
+        // next parser is not present, so we have finished parsing a message
+        MessageSharedPtr message = result.message_;
+        ENVOY_LOG(trace, "parsed message: {}", *message);
+        for (auto& listener : listeners_) {
+          listener->onMessage(result.message_);
+        }
+
+        // we finished parsing this request, start anew
+        parser = std::make_shared<RequestStartParser>(parser_resolver_);
+      } else {
+        parser = result.next_parser_;
+      }
+      result = parser->parse(buffer, remaining);
+    }
+  }
+}
+
+void ResponseDecoder::onWrite(Buffer::Instance&) { /* not implemented yet */
+}
+
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy
diff --git a/source/extensions/filters/network/kafka/codec.h b/source/extensions/filters/network/kafka/codec.h
new file mode 100644
index 0000000000000..2f7d5597ec94b
--- /dev/null
+++ b/source/extensions/filters/network/kafka/codec.h
@@ -0,0 +1,49 @@
+#pragma once
+
+#include "envoy/buffer/buffer.h"
+#include "envoy/common/pure.h"
+
+#include "extensions/filters/network/kafka/kafka_request.h"
+#include "extensions/filters/network/kafka/parser.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+
+class MessageListener {
+public:
+  virtual ~MessageListener(){};
+
+  virtual void onMessage(MessageSharedPtr) PURE;
+};
+
+typedef std::shared_ptr<MessageListener> MessageListenerPtr;
+
+class RequestDecoder : public Logger::Loggable<Logger::Id::kafka> {
+public:
+  RequestDecoder(const RequestParserResolver parserResolver,
+                 const std::vector<MessageListenerPtr> listeners)
+      : parser_resolver_{parserResolver}, listeners_{listeners},
+        request_parser_{new RequestStartParser(parser_resolver_)} {};
+
+  void onData(Buffer::Instance& data);
+
+private:
+  void doParse(ParserSharedPtr& parser, const Buffer::RawSlice& slice);
+
+  const RequestParserResolver parser_resolver_;
+  const std::vector<MessageListenerPtr> listeners_;
+
+  ParserSharedPtr request_parser_;
+};
+
+class ResponseDecoder {
+public:
+  void onWrite(Buffer::Instance& data);
+};
+
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy
diff --git a/source/extensions/filters/network/kafka/config.cc b/source/extensions/filters/network/kafka/config.cc
new file mode 100644
index 0000000000000..1acad0182f449
--- /dev/null
+++ b/source/extensions/filters/network/kafka/config.cc
@@ -0,0 +1,57 @@
+#include "envoy/registry/registry.h"
+#include "envoy/server/filter_config.h"
+#include "envoy/stats/scope.h"
+
+#include "extensions/filters/network/kafka/kafka_filter.h"
+#include "extensions/filters/network/kafka/metrics_holder.h"
+#include "extensions/filters/network/well_known_names.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+
+/**
+ * Config registration for the Kafka filter. @see NamedNetworkFilterConfigFactory.
+ */
+class KafkaConfigFactory : public Server::Configuration::NamedNetworkFilterConfigFactory {
+public:
+  // NamedNetworkFilterConfigFactory
+  Network::FilterFactoryCb
+  createFilterFactory(const Json::Object&,
+                      Server::Configuration::FactoryContext& context) override {
+    return createInternal(context.scope());
+  }
+
+  Network::FilterFactoryCb
+  createFilterFactoryFromProto(const Protobuf::Message&,
+                               Server::Configuration::FactoryContext& context) override {
+    return createInternal(context.scope());
+  }
+
+  ProtobufTypes::MessagePtr createEmptyConfigProto() override {
+    return ProtobufTypes::MessagePtr{new Envoy::ProtobufWkt::Empty()};
+  }
+
+  std::string name() override { return NetworkFilterNames::get().Kafka; }
+
+private:
+  Network::FilterFactoryCb createInternal(Stats::Scope& scope) {
+    std::shared_ptr<MetricsHolder> metrics_holder = std::make_shared<MetricsHolder>(scope);
+    return [&scope, metrics_holder](Network::FilterManager& filter_manager) -> void {
+      filter_manager.addFilter(std::make_shared<KafkaFilter>(scope, metrics_holder));
+    };
+  }
+};
+
+/**
+ * Static registration for the Kafka filter. @see RegisterFactory.
+ */
+static Registry::RegisterFactory<KafkaConfigFactory,
+                                 Server::Configuration::NamedNetworkFilterConfigFactory>
+    registered_;
+
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy
diff --git a/source/extensions/filters/network/kafka/kafka_filter.cc b/source/extensions/filters/network/kafka/kafka_filter.cc
new file mode 100644
index 0000000000000..41313331f5619
--- /dev/null
+++ b/source/extensions/filters/network/kafka/kafka_filter.cc
@@ -0,0 +1,49 @@
+#include "extensions/filters/network/kafka/kafka_filter.h"
+
+#include <sstream>
+
+#include "envoy/buffer/buffer.h"
+#include "envoy/network/connection.h"
+
+#include "common/common/assert.h"
+
+#include "extensions/filters/network/kafka/kafka_request.h"
+#include "extensions/filters/network/kafka/kafka_response.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+
+class LoggingMessageListener : public MessageListener, public Logger::Loggable<Logger::Id::kafka> {
+public:
+  void onMessage(MessageSharedPtr arg) override { ENVOY_LOG(info, "received: {}", *arg); }
+};
+
+KafkaFilter::KafkaFilter(Stats::Scope& scope, std::shared_ptr<MetricsHolder> metrics_holder)
+    : stats_{KAFKA_STATS(POOL_COUNTER_PREFIX(scope, "kafka."))}, metrics_holder_{metrics_holder},
+      request_decoder_{RequestParserResolver::KAFKA_0_11,
+                       {std::make_unique<LoggingMessageListener>()}} {
+  stats_.filters_created_.inc();
+};
+
+Network::FilterStatus KafkaFilter::onNewConnection() { return Network::FilterStatus::Continue; }
+
+void KafkaFilter::initializeReadFilterCallbacks(Network::ReadFilterCallbacks& callbacks) {
+  read_callbacks_ = &callbacks;
+}
+
+Network::FilterStatus KafkaFilter::onData(Buffer::Instance& data, bool) {
+  request_decoder_.onData(data);
+  return Network::FilterStatus::Continue;
+}
+
+Network::FilterStatus KafkaFilter::onWrite(Buffer::Instance& data, bool) {
+  response_decoder_.onWrite(data);
+  return Network::FilterStatus::Continue;
+}
+
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy
diff --git a/source/extensions/filters/network/kafka/kafka_filter.h b/source/extensions/filters/network/kafka/kafka_filter.h
new file mode 100644
index 0000000000000..c57cd4c793070
--- /dev/null
+++ b/source/extensions/filters/network/kafka/kafka_filter.h
@@ -0,0 +1,72 @@
+#pragma once
+
+#include <sstream>
+
+#include "envoy/network/filter.h"
+#include "envoy/stats/scope.h"
+
+#include "common/common/logger.h"
+
+#include "extensions/filters/network/kafka/codec.h"
+#include "extensions/filters/network/kafka/kafka_request.h"
+#include "extensions/filters/network/kafka/kafka_response.h"
+#include "extensions/filters/network/kafka/metrics_holder.h"
+#include "extensions/filters/network/kafka/parser.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+
+// === STATS ===================================================================
+
+/**
+ * All Kafka stats. @see stats_macros.h
+ */
+// clang-format off
+#define KAFKA_STATS(COUNTER)                                                                \
+  COUNTER(filters_created)                                                                  \
+// clang-format on
+
+/**
+ * Struct definition for all Kafka stats. @see stats_macros.h
+ */
+struct InstanceStats {
+  KAFKA_STATS(GENERATE_COUNTER_STRUCT)
+};
+
+// === FILTER ==================================================================
+
+/**
+ * Implementation of a basic kafka filter.
+ */
+class KafkaFilter : public Network::Filter {
+
+public:
+  KafkaFilter(Stats::Scope& scope, std::shared_ptr<MetricsHolder>);
+  virtual ~KafkaFilter() {}
+
+  // Network::ReadFilter
+  Network::FilterStatus onNewConnection() override;
+  void initializeReadFilterCallbacks(Network::ReadFilterCallbacks& callbacks) override;
+  Network::FilterStatus onData(Buffer::Instance& data, bool end_stream) override;
+
+  // Network::WriteFilter
+  Network::FilterStatus onWrite(Buffer::Instance& data, bool end_stream) override;
+
+private:
+  Network::ReadFilterCallbacks* read_callbacks_{};
+
+  InstanceStats stats_;
+  std::shared_ptr<MetricsHolder> metrics_holder_;
+
+  // in future, we might have some shared state shared between both request & response decoder
+  // for e.g. matching responses with requests (by correlation id)
+  RequestDecoder request_decoder_;
+  ResponseDecoder response_decoder_;
+};
+
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy
diff --git a/source/extensions/filters/network/kafka/kafka_protocol.h b/source/extensions/filters/network/kafka/kafka_protocol.h
new file mode 100644
index 0000000000000..f6452abd3e859
--- /dev/null
+++ b/source/extensions/filters/network/kafka/kafka_protocol.h
@@ -0,0 +1,123 @@
+#pragma once
+
+#include <vector>
+
+#include "common/common/macros.h"
+
+#include "extensions/filters/network/kafka/kafka_types.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+
+// from http://kafka.apache.org/protocol.html#protocol_api_keys
+enum RequestType : INT16 {
+  Produce = 0,
+  Fetch = 1,
+  ListOffsets = 2,
+  Metadata = 3,
+  LeaderAndIsr = 4,
+  StopReplica = 5,
+  UpdateMetadata = 6,
+  ControlledShutdown = 7,
+  OffsetCommit = 8,
+  OffsetFetch = 9,
+  FindCoordinator = 10,
+  JoinGroup = 11,
+  Heartbeat = 12,
+  LeaveGroup = 13,
+  SyncGroup = 14,
+  DescribeGroups = 15,
+  ListGroups = 16,
+  SaslHandshake = 17,
+  ApiVersions = 18,
+  CreateTopics = 19,
+  DeleteTopics = 20,
+  DeleteRecords = 21,
+  InitProducerId = 22,
+  OffsetForLeaderEpoch = 23,
+  AddPartitionsToTxn = 24,
+  AddOffsetsToTxn = 25,
+  EndTxn = 26,
+  WriteTxnMarkers = 27,
+  TxnOffsetCommit = 28,
+  DescribeAcls = 29,
+  CreateAcls = 30,
+  DeleteAcls = 31,
+  DescribeConfigs = 32,
+  AlterConfigs = 33,
+  AlterReplicaLogDirs = 34,
+  DescribeLogDirs = 35,
+  SaslAuthenticate = 36,
+  CreatePartitions = 37,
+  CreateDelegationToken = 38,
+  RenewDelegationToken = 39,
+  ExpireDelegationToken = 40,
+  DescribeDelegationToken = 41,
+  DeleteGroups = 42
+};
+
+struct RequestSpec {
+  const INT16 api_key_;
+  const std::string name_;
+};
+
+struct KafkaRequest {
+
+  // clang-format off
+  static const std::vector<RequestSpec>& requests() {
+    CONSTRUCT_ON_FIRST_USE(
+        std::vector<RequestSpec>,
+        {RequestType::Produce, "Produce"},
+        {RequestType::Fetch, "Fetch"},
+        {RequestType::ListOffsets, "ListOffsets"},
+        {RequestType::Metadata, "Metadata"},
+        {RequestType::LeaderAndIsr, "LeaderAndIsr"},
+        {RequestType::StopReplica, "StopReplica"},
+        {RequestType::UpdateMetadata, "UpdateMetadata"},
+        {RequestType::ControlledShutdown, "ControlledShutdown"},
+        {RequestType::OffsetCommit, "OffsetCommit"},
+        {RequestType::OffsetFetch, "OffsetFetch"},
+        {RequestType::FindCoordinator, "FindCoordinator"},
+        {RequestType::JoinGroup, "JoinGroup"},
+        {RequestType::Heartbeat, "Heartbeat"},
+        {RequestType::LeaveGroup, "LeaveGroup"},
+        {RequestType::SyncGroup, "SyncGroup"},
+        {RequestType::DescribeGroups, "DescribeGroups"},
+        {RequestType::ListGroups, "ListGroups"},
+        {RequestType::SaslHandshake, "SaslHandshake"},
+        {RequestType::ApiVersions, "ApiVersions"},
+        {RequestType::CreateTopics, "CreateTopics"},
+        {RequestType::DeleteTopics, "DeleteTopics"},
+        {RequestType::DeleteRecords, "DeleteRecords"},
+        {RequestType::InitProducerId, "InitProducerId"},
+        {RequestType::OffsetForLeaderEpoch, "OffsetForLeaderEpoch"},
+        {RequestType::AddPartitionsToTxn, "AddPartitionsToTxn"},
+        {RequestType::AddOffsetsToTxn, "AddOffsetsToTxn"},
+        {RequestType::EndTxn, "EndTxn"},
+        {RequestType::WriteTxnMarkers, "WriteTxnMarkers"},
+        {RequestType::TxnOffsetCommit, "TxnOffsetCommit"},
+        {RequestType::DescribeAcls, "DescribeAcls"},
+        {RequestType::CreateAcls, "CreateAcls"},
+        {RequestType::DeleteAcls, "DeleteAcls"},
+        {RequestType::DescribeConfigs, "DescribeConfigs"},
+        {RequestType::AlterConfigs, "AlterConfigs"},
+        {RequestType::AlterReplicaLogDirs, "AlterReplicaLogDirs"},
+        {RequestType::DescribeLogDirs, "DescribeLogDirs"},
+        {RequestType::SaslAuthenticate, "SaslAuthenticate"},
+        {RequestType::CreatePartitions, "CreatePartitions"},
+        {RequestType::CreateDelegationToken, "CreateDelegationToken"},
+        {RequestType::RenewDelegationToken, "RenewDelegationToken"},
+        {RequestType::ExpireDelegationToken, "ExpireDelegationToken"},
+        {RequestType::DescribeDelegationToken, "DescribeDelegationToken"},
+        {RequestType::DeleteGroups, "DeleteGroups"}
+    );
+  }
+  // clang-format on
+};
+
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy
diff --git a/source/extensions/filters/network/kafka/kafka_request.cc b/source/extensions/filters/network/kafka/kafka_request.cc
new file mode 100644
index 0000000000000..5b298aa1829f7
--- /dev/null
+++ b/source/extensions/filters/network/kafka/kafka_request.cc
@@ -0,0 +1,131 @@
+#include "extensions/filters/network/kafka/kafka_request.h"
+
+#include "extensions/filters/network/kafka/kafka_protocol.h"
+#include "extensions/filters/network/kafka/parser.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+
+// === REQUEST PARSER MAPPING (REQUEST TYPE => PARSER) =========================
+
+GeneratorMap computeGeneratorMap(std::vector<ParserSpec> specs) {
+  GeneratorMap result;
+  for (auto& spec : specs) {
+    auto generators = result[spec.api_key_];
+    if (!generators) {
+      generators = std::make_shared<std::unordered_map<INT16, GeneratorFunction>>();
+      result[spec.api_key_] = generators;
+    }
+    for (INT16 api_version : spec.api_versions_) {
+      (*generators)[api_version] = spec.generator_;
+    }
+  }
+
+  return result;
+}
+
+#define PARSER_SPEC(REQUEST_NAME, PARSER_VERSION, ...)                                             \
+  ParserSpec {                                                                                     \
+    RequestType::REQUEST_NAME, {__VA_ARGS__}, [](RequestContextSharedPtr arg) -> ParserSharedPtr { \
+      return std::make_shared<REQUEST_NAME##Request##PARSER_VERSION##Parser>(arg);                 \
+    }                                                                                              \
+  }
+
+const RequestParserResolver RequestParserResolver::KAFKA_0_11{{
+    PARSER_SPEC(Produce, V0, 0, 1, 2),
+    PARSER_SPEC(Produce, V3, 3),
+    PARSER_SPEC(Fetch, V0, 0, 1, 2),
+    PARSER_SPEC(Fetch, V3, 3),
+    PARSER_SPEC(Fetch, V4, 4),
+    PARSER_SPEC(Fetch, V5, 5),
+    PARSER_SPEC(ListOffsets, V0, 0),
+    PARSER_SPEC(ListOffsets, V1, 1),
+    PARSER_SPEC(ListOffsets, V2, 2),
+    PARSER_SPEC(Metadata, V0, 0, 1, 2, 3),
+    PARSER_SPEC(Metadata, V4, 4),
+    // XXX(adam.kotwasinski) missing request types here
+    PARSER_SPEC(OffsetCommit, V0, 0),
+    PARSER_SPEC(OffsetCommit, V1, 1),
+    PARSER_SPEC(OffsetCommit, V2, 2, 3),
+    PARSER_SPEC(OffsetFetch, V0, 0, 1),
+    PARSER_SPEC(OffsetFetch, V2, 2, 3),
+    PARSER_SPEC(ApiVersions, V0, 0, 1),
+}};
+
+ParserSharedPtr RequestParserResolver::createParser(INT16 api_key, INT16 api_version,
+                                                    RequestContextSharedPtr context) const {
+  const auto api_versions_ptr = generators_.find(api_key);
+  // unknown api_key
+  if (generators_.end() == api_versions_ptr) {
+    return std::make_shared<SentinelConsumer>(context);
+  }
+  const auto api_versions = api_versions_ptr->second;
+
+  // unknown api_version
+  const auto generator = api_versions->find(api_version);
+  if (api_versions->end() == generator) {
+    return std::make_shared<SentinelConsumer>(context);
+  }
+
+  // found matching parser generator, create parser
+  return generator->second(context);
+}
+
+// === HEADER PARSERS ==========================================================
+
+ParseResponse RequestStartParser::parse(const char*& buffer, uint64_t& remaining) {
+  buffer_.feed(buffer, remaining);
+  if (buffer_.ready()) {
+    context_->remaining_request_size_ = buffer_.get();
+    return ParseResponse::nextParser(
+        std::make_shared<RequestHeaderParser>(parser_resolver_, context_));
+  } else {
+    return ParseResponse::stillWaiting();
+  }
+}
+
+ParseResponse RequestHeaderParser::parse(const char*& buffer, uint64_t& remaining) {
+  context_->remaining_request_size_ -= buffer_.feed(buffer, remaining);
+
+  if (buffer_.ready()) {
+    RequestHeader request_header = buffer_.get();
+    context_->request_header_ = request_header;
+    ParserSharedPtr next_parser = parser_resolver_.createParser(
+        request_header.api_key_, request_header.api_version_, context_);
+    return ParseResponse::nextParser(next_parser);
+  } else {
+    return ParseResponse::stillWaiting();
+  }
+}
+
+// === UNKNOWN REQUEST =========================================================
+
+ParseResponse SentinelConsumer::parse(const char*& buffer, uint64_t& remaining) {
+  const size_t min = std::min<size_t>(context_->remaining_request_size_, remaining);
+  buffer += min;
+  remaining -= min;
+  context_->remaining_request_size_ -= min;
+  if (0 == context_->remaining_request_size_) {
+    return ParseResponse::parsedMessage(
+        std::make_shared<UnknownRequest>(context_->request_header_));
+  } else {
+    return ParseResponse::stillWaiting();
+  }
+}
+
+// === REQUEST SERIALIZER ======================================================
+
+size_t RequestSerializer::encode(const Request& request, char* dst) {
+  Encoder encoder;
+  // encode payload first, then the payload's length
+  INT32 data_len = encoder.encode(request, dst + sizeof(INT32));
+  INT32 header_len = encoder.encode(data_len, dst);
+  return header_len + data_len;
+}
+
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy
diff --git a/source/extensions/filters/network/kafka/kafka_request.h b/source/extensions/filters/network/kafka/kafka_request.h
new file mode 100644
index 0000000000000..fad49079b0f40
--- /dev/null
+++ b/source/extensions/filters/network/kafka/kafka_request.h
@@ -0,0 +1,857 @@
+#pragma once
+
+#include <sstream>
+
+#include "envoy/common/exception.h"
+
+#include "common/common/assert.h"
+
+#include "extensions/filters/network/kafka/kafka_protocol.h"
+#include "extensions/filters/network/kafka/parser.h"
+#include "extensions/filters/network/kafka/serialization.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+
+// === VECTOR ==================================================================
+
+template <typename T> std::ostream& operator<<(std::ostream& os, const std::vector<T>& arg) {
+  os << "[";
+  for (auto iter = arg.begin(); iter != arg.end(); iter++) {
+    if (iter != arg.begin()) {
+      os << ", ";
+    }
+    os << *iter;
+  }
+  os << "]";
+  return os;
+}
+
+template <typename T> std::ostream& operator<<(std::ostream& os, const absl::optional<T>& arg) {
+  if (arg.has_value()) {
+    os << *arg;
+  } else {
+    os << "<null>";
+  }
+  return os;
+}
+
+// === REQUEST =================================================================
+
+struct RequestHeader {
+  INT16 api_key_;
+  INT16 api_version_;
+  INT32 correlation_id_;
+  NULLABLE_STRING client_id_;
+
+  bool operator==(const RequestHeader& rhs) const {
+    return api_key_ == rhs.api_key_ && api_version_ == rhs.api_version_ &&
+           correlation_id_ == rhs.correlation_id_ && client_id_ == rhs.client_id_;
+  };
+
+  friend std::ostream& operator<<(std::ostream& os, const RequestHeader& arg) {
+    return os << "{api_key = " << arg.api_key_ << ", api_version = " << arg.api_version_
+              << ", correlation_id = " << arg.correlation_id_ << ", client_id = " << arg.client_id_
+              << "}";
+  };
+};
+
+struct RequestContext {
+  INT32 remaining_request_size_{0};
+  RequestHeader request_header_{};
+
+  friend std::ostream& operator<<(std::ostream& os, const RequestContext& arg) {
+    return os << "{header = " << arg.request_header_
+              << ", remaining = " << arg.remaining_request_size_ << "}";
+  }
+};
+
+typedef std::shared_ptr<RequestContext> RequestContextSharedPtr;
+
+// === REQUEST PARSER MAPPING (REQUEST TYPE => PARSER) =========================
+
+// a function generating a parser with given context
+typedef std::function<ParserSharedPtr(RequestContextSharedPtr)> GeneratorFunction;
+
+// two-level map: api_key -> api_version -> generator function
+typedef std::unordered_map<INT16, std::shared_ptr<std::unordered_map<INT16, GeneratorFunction>>>
+    GeneratorMap;
+
+struct ParserSpec {
+  const INT16 api_key_;
+  const std::vector<INT16> api_versions_;
+  const GeneratorFunction generator_;
+};
+
+// helper function that generates a map from specs looking like { api_key, api_versions... }
+GeneratorMap computeGeneratorMap(std::vector<ParserSpec> arg);
+
+/**
+ * Responsible for mapping [apiKey, apiVersion] -> request parser
+ */
+class RequestParserResolver {
+public:
+  RequestParserResolver(std::vector<ParserSpec> arg) : generators_{computeGeneratorMap(arg)} {};
+
+  ParserSharedPtr createParser(INT16 api_key, INT16 api_version,
+                               RequestContextSharedPtr context) const;
+
+  static const RequestParserResolver KAFKA_0_11;
+
+private:
+  GeneratorMap generators_;
+};
+
+// === HEADER PARSERS ==========================================================
+
+class RequestStartParser : public Parser {
+public:
+  RequestStartParser(const RequestParserResolver& parser_resolver)
+      : parser_resolver_{parser_resolver}, context_{std::make_shared<RequestContext>()} {};
+
+  ParseResponse parse(const char*& buffer, uint64_t& remaining);
+
+private:
+  const RequestParserResolver& parser_resolver_;
+  const RequestContextSharedPtr context_;
+  Int32Buffer buffer_;
+};
+
+class RequestHeaderBuffer : public CompositeBuffer<RequestHeader, Int16Buffer, Int16Buffer,
+                                                   Int32Buffer, NullableStringBuffer> {};
+
+class RequestHeaderParser : public Parser {
+public:
+  RequestHeaderParser(const RequestParserResolver& parser_resolver, RequestContextSharedPtr context)
+      : parser_resolver_{parser_resolver}, context_{context} {};
+
+  ParseResponse parse(const char*& buffer, uint64_t& remaining);
+
+private:
+  const RequestParserResolver& parser_resolver_;
+  const RequestContextSharedPtr context_;
+  RequestHeaderBuffer buffer_;
+};
+
+// === BASE REQUEST PARSER =====================================================
+
+template <typename RT, typename BT> class BufferedParser : public Parser {
+public:
+  BufferedParser(RequestContextSharedPtr ctx) : ctx_{ctx} {};
+  ParseResponse parse(const char*& buffer, uint64_t& remaining) override;
+
+protected:
+  RequestContextSharedPtr ctx_;
+  BT buffer_;
+};
+
+template <typename RT, typename BT>
+ParseResponse BufferedParser<RT, BT>::parse(const char*& buffer, uint64_t& remaining) {
+  ctx_->remaining_request_size_ -= buffer_.feed(buffer, remaining);
+  if (buffer_.ready()) {
+    // after a successful parse, there should be nothing left
+    ASSERT(0 == ctx_->remaining_request_size_);
+    RT request = buffer_.get();
+    request.header() = ctx_->request_header_;
+    ENVOY_LOG(trace, "parsed request {}: {}", *ctx_, request);
+    MessageSharedPtr msg = std::make_shared<RT>(request);
+    return ParseResponse::parsedMessage(msg);
+  } else {
+    return ParseResponse::stillWaiting();
+  }
+}
+
+// names of Buffers/Parsers are influenced by org.apache.kafka.common.protocol.Protocol names
+
+// clang-format off
+#define DEFINE_REQUEST_PARSER(REQUEST_TYPE, VERSION) \
+class REQUEST_TYPE ## VERSION ## Parser :                                      \
+  public BufferedParser< REQUEST_TYPE , REQUEST_TYPE ## VERSION ## Buffer >{   \
+public:                                                                        \
+  REQUEST_TYPE ## VERSION ## Parser(RequestContextSharedPtr ctx):              \
+    BufferedParser{ctx} {};                                                    \
+};
+// clang-format on
+
+// === REQUEST BASE ============================================================
+
+class Request : public Message {
+public:
+  /**
+   * Request header fields need to be initialized by user in case of newly created requests
+   */
+  Request(INT16 api_key) : request_header_{api_key, 0, 0, ""} {};
+
+  Request(const RequestHeader& request_header) : request_header_{request_header} {};
+
+  RequestHeader& header() { return request_header_; }
+
+  INT16& apiVersion() { return request_header_.api_version_; }
+
+  INT32& correlationId() { return request_header_.correlation_id_; }
+
+  NULLABLE_STRING& clientId() { return request_header_.client_id_; }
+
+  size_t encode(char* dst, Encoder& encoder) const {
+    size_t written{0};
+    written += encoder.encode(request_header_.api_key_, dst + written);
+    written += encoder.encode(request_header_.api_version_, dst + written);
+    written += encoder.encode(request_header_.correlation_id_, dst + written);
+    written += encoder.encode(request_header_.client_id_, dst + written);
+    written += encodeDetails(dst + written, encoder);
+    return written;
+  }
+
+  std::ostream& print(std::ostream& os) const override final {
+    os << request_header_ << " "; // not very pretty
+    return printDetails(os);
+  }
+
+protected:
+  virtual size_t encodeDetails(char*, Encoder&) const PURE;
+
+  virtual std::ostream& printDetails(std::ostream& os) const PURE;
+
+  RequestHeader request_header_;
+};
+
+// === PRODUCE (0) =============================================================
+
+struct ProducePartition {
+  INT32 partition_;
+  INT32 data_size_;
+
+  friend std::ostream& operator<<(std::ostream& os, const ProducePartition& arg) {
+    return os << "{partition = " << arg.partition_ << ", data_size = " << arg.data_size_ << "}";
+  }
+};
+
+struct ProduceTopic {
+  STRING topic_;
+  NULLABLE_ARRAY<ProducePartition> partitions_;
+
+  friend std::ostream& operator<<(std::ostream& os, const ProduceTopic& arg) {
+    return os << "{topic = " << arg.topic_ << ", partitions = " << arg.partitions_ << "}";
+  }
+};
+
+class ProduceRequest : public Request {
+public:
+  // v0 .. v2
+  ProduceRequest(INT16 acks, INT32 timeout, NULLABLE_ARRAY<ProduceTopic> topics)
+      : ProduceRequest("no-tx-id", acks, timeout, topics){};
+
+  // v3
+  ProduceRequest(NULLABLE_STRING transactional_id, INT16 acks, INT32 timeout,
+                 NULLABLE_ARRAY<ProduceTopic> topics)
+      : Request{RequestType::Produce},
+        transactional_id_{transactional_id}, acks_{acks}, timeout_{timeout}, topics_{topics} {};
+
+protected:
+  size_t encodeDetails(char*, Encoder&) const override { throw EnvoyException("not implemented"); }
+
+  std::ostream& printDetails(std::ostream& os) const override {
+    return os << "{transactional_id = " << transactional_id_ << ", acks = " << acks_
+              << ", timeout = " << timeout_ << ", topics = " << topics_ << "}";
+  };
+
+private:
+  NULLABLE_STRING transactional_id_;
+  INT16 acks_;
+  INT32 timeout_;
+  NULLABLE_ARRAY<ProduceTopic> topics_;
+};
+
+class ProducePartitionBuffer
+    : public CompositeBuffer<ProducePartition, Int32Buffer, NullableBytesIgnoringBuffer> {};
+class ProducePartitionArrayBuffer : public ArrayBuffer<ProducePartition, ProducePartitionBuffer> {};
+class ProduceTopicBuffer
+    : public CompositeBuffer<ProduceTopic, StringBuffer, ProducePartitionArrayBuffer> {};
+class ProduceTopicArrayBuffer : public ArrayBuffer<ProduceTopic, ProduceTopicBuffer> {};
+
+class ProduceRequestV0Buffer
+    : public CompositeBuffer<ProduceRequest, Int16Buffer, Int32Buffer, ProduceTopicArrayBuffer> {};
+class ProduceRequestV3Buffer
+    : public CompositeBuffer<ProduceRequest, NullableStringBuffer, Int16Buffer, Int32Buffer,
+                             ProduceTopicArrayBuffer> {};
+
+DEFINE_REQUEST_PARSER(ProduceRequest, V0);
+DEFINE_REQUEST_PARSER(ProduceRequest, V3);
+
+// === FETCH (1) ===============================================================
+
+struct FetchRequestPartition {
+  INT32 partition_;
+  INT64 fetch_offset_;
+  INT64 log_start_offset_;
+  INT32 max_bytes_;
+
+  size_t encode(char* dst, Encoder& encoder) const {
+    size_t written{0};
+    written += encoder.encode(partition_, dst + written);
+    written += encoder.encode(fetch_offset_, dst + written);
+    written += encoder.encode(log_start_offset_, dst + written);
+    written += encoder.encode(max_bytes_, dst + written);
+    return written;
+  }
+
+  friend std::ostream& operator<<(std::ostream& os, const FetchRequestPartition& arg) {
+    return os << "{partition = " << arg.partition_ << ", fetch_offset = " << arg.fetch_offset_
+              << ", log_start_offset = " << arg.log_start_offset_
+              << ", max_bytes = " << arg.max_bytes_ << "}";
+  }
+
+  bool operator==(const FetchRequestPartition& rhs) const {
+    return partition_ == rhs.partition_ && fetch_offset_ == rhs.fetch_offset_ &&
+           log_start_offset_ == rhs.log_start_offset_ && max_bytes_ == rhs.max_bytes_;
+  };
+
+  // v0 .. v4
+  FetchRequestPartition(INT32 partition, INT64 fetch_offset, INT32 max_bytes)
+      : FetchRequestPartition(partition, fetch_offset, -1, max_bytes){};
+
+  // v5
+  FetchRequestPartition(INT32 partition, INT64 fetch_offset, INT64 log_start_offset,
+                        INT32 max_bytes)
+      : partition_{partition}, fetch_offset_{fetch_offset}, log_start_offset_{log_start_offset},
+        max_bytes_{max_bytes} {};
+};
+
+struct FetchRequestTopic {
+  STRING topic_;
+  NULLABLE_ARRAY<FetchRequestPartition> partitions_;
+
+  size_t encode(char* dst, Encoder& encoder) const {
+    size_t written{0};
+    written += encoder.encode(topic_, dst + written);
+    written += encoder.encode(partitions_, dst + written);
+    return written;
+  }
+
+  bool operator==(const FetchRequestTopic& rhs) const {
+    return topic_ == rhs.topic_ && partitions_ == rhs.partitions_;
+  };
+
+  friend std::ostream& operator<<(std::ostream& os, const FetchRequestTopic& arg) {
+    return os << "{topic = " << arg.topic_ << ", partitions = " << arg.partitions_ << "}";
+  }
+};
+
+class FetchRequest : public Request {
+public:
+  // v0 .. v2
+  FetchRequest(INT32 replica_id, INT32 max_wait_time, INT32 min_bytes,
+               NULLABLE_ARRAY<FetchRequestTopic> topics)
+      : FetchRequest(replica_id, max_wait_time, min_bytes, -1, topics){};
+
+  // v3
+  FetchRequest(INT32 replica_id, INT32 max_wait_time, INT32 min_bytes, INT32 max_bytes,
+               NULLABLE_ARRAY<FetchRequestTopic> topics)
+      : FetchRequest(replica_id, max_wait_time, min_bytes, max_bytes, -1, topics){};
+
+  // v4 .. v5
+  FetchRequest(INT32 replica_id, INT32 max_wait_time, INT32 min_bytes, INT32 max_bytes,
+               INT8 isolation_level, NULLABLE_ARRAY<FetchRequestTopic> topics)
+      : Request{RequestType::Fetch}, replica_id_{replica_id}, max_wait_time_{max_wait_time},
+        min_bytes_{min_bytes}, max_bytes_{max_bytes},
+        isolation_level_{isolation_level}, topics_{topics} {};
+
+  bool operator==(const FetchRequest& rhs) const {
+    return request_header_ == rhs.request_header_ && replica_id_ == rhs.replica_id_ &&
+           max_wait_time_ == rhs.max_wait_time_ && min_bytes_ == rhs.min_bytes_ &&
+           max_bytes_ == rhs.max_bytes_ && isolation_level_ == rhs.isolation_level_ &&
+           topics_ == rhs.topics_;
+  };
+
+protected:
+  size_t encodeDetails(char* dst, Encoder& encoder) const override {
+    size_t written{0};
+    written += encoder.encode(replica_id_, dst + written);
+    written += encoder.encode(max_wait_time_, dst + written);
+    written += encoder.encode(min_bytes_, dst + written);
+    written += encoder.encode(max_bytes_, dst + written);
+    written += encoder.encode(isolation_level_, dst + written);
+    written += encoder.encode(topics_, dst + written);
+    return written;
+  }
+
+  std::ostream& printDetails(std::ostream& os) const override {
+    return os << "{replica_id = " << replica_id_ << ", max_wait_time = " << max_wait_time_
+              << ", min_bytes = " << min_bytes_ << ", max_bytes = " << max_bytes_
+              << ", isolation_level = " << static_cast<uint32_t>(isolation_level_)
+              << ", topics = " << topics_ << "}";
+  }
+
+private:
+  INT32 replica_id_;
+  INT32 max_wait_time_;
+  INT32 min_bytes_;
+  INT32 max_bytes_;
+  INT8 isolation_level_;
+  NULLABLE_ARRAY<FetchRequestTopic> topics_;
+};
+
+class FetchRequestPartitionV0Buffer
+    : public CompositeBuffer<FetchRequestPartition, Int32Buffer, Int64Buffer, Int32Buffer> {};
+class FetchRequestPartitionV0ArrayBuffer
+    : public ArrayBuffer<FetchRequestPartition, FetchRequestPartitionV0Buffer> {};
+class FetchRequestTopicV0Buffer
+    : public CompositeBuffer<FetchRequestTopic, StringBuffer, FetchRequestPartitionV0ArrayBuffer> {
+};
+class FetchRequestTopicV0ArrayBuffer
+    : public ArrayBuffer<FetchRequestTopic, FetchRequestTopicV0Buffer> {};
+
+class FetchRequestPartitionV5Buffer
+    : public CompositeBuffer<FetchRequestPartition, Int32Buffer, Int64Buffer, Int64Buffer,
+                             Int32Buffer> {};
+class FetchRequestPartitionV5ArrayBuffer
+    : public ArrayBuffer<FetchRequestPartition, FetchRequestPartitionV5Buffer> {};
+class FetchRequestTopicV5Buffer
+    : public CompositeBuffer<FetchRequestTopic, StringBuffer, FetchRequestPartitionV5ArrayBuffer> {
+};
+class FetchRequestTopicV5ArrayBuffer
+    : public ArrayBuffer<FetchRequestTopic, FetchRequestTopicV5Buffer> {};
+
+class FetchRequestV0Buffer : public CompositeBuffer<FetchRequest, Int32Buffer, Int32Buffer,
+                                                    Int32Buffer, FetchRequestTopicV0ArrayBuffer> {};
+class FetchRequestV3Buffer
+    : public CompositeBuffer<FetchRequest, Int32Buffer, Int32Buffer, Int32Buffer, Int32Buffer,
+                             FetchRequestTopicV0ArrayBuffer> {};
+class FetchRequestV4Buffer
+    : public CompositeBuffer<FetchRequest, Int32Buffer, Int32Buffer, Int32Buffer, Int32Buffer,
+                             Int8Buffer, FetchRequestTopicV0ArrayBuffer> {};
+class FetchRequestV5Buffer
+    : public CompositeBuffer<FetchRequest, Int32Buffer, Int32Buffer, Int32Buffer, Int32Buffer,
+                             Int8Buffer, FetchRequestTopicV5ArrayBuffer> {};
+
+DEFINE_REQUEST_PARSER(FetchRequest, V0);
+DEFINE_REQUEST_PARSER(FetchRequest, V3);
+DEFINE_REQUEST_PARSER(FetchRequest, V4);
+DEFINE_REQUEST_PARSER(FetchRequest, V5);
+
+// === LIST OFFSETS (2) ========================================================
+
+struct ListOffsetsPartition {
+  INT32 partition_;
+  INT64 timestamp_;
+
+  size_t encode(char* dst, Encoder& encoder) const {
+    size_t written{0};
+    written += encoder.encode(partition_, dst + written);
+    written += encoder.encode(timestamp_, dst + written);
+    return written;
+  }
+
+  bool operator==(const ListOffsetsPartition& rhs) const {
+    return partition_ == rhs.partition_ && timestamp_ == rhs.timestamp_;
+  };
+
+  friend std::ostream& operator<<(std::ostream& os, const ListOffsetsPartition& arg) {
+    return os << "{partition = " << arg.partition_ << ", timestamp = " << arg.timestamp_ << "}";
+  }
+
+  // v0
+  ListOffsetsPartition(INT32 partition, INT64 timestamp, INT32 /* (ignored) max_num_offsets */)
+      : ListOffsetsPartition(partition, timestamp){};
+
+  // v1 .. v2
+  ListOffsetsPartition(INT32 partition, INT64 timestamp)
+      : partition_{partition}, timestamp_{timestamp} {};
+};
+
+struct ListOffsetsTopic {
+  STRING topic_;
+  NULLABLE_ARRAY<ListOffsetsPartition> partitions_;
+
+  size_t encode(char* dst, Encoder& encoder) const {
+    size_t written{0};
+    written += encoder.encode(topic_, dst + written);
+    written += encoder.encode(partitions_, dst + written);
+    return written;
+  }
+
+  bool operator==(const ListOffsetsTopic& rhs) const {
+    return topic_ == rhs.topic_ && partitions_ == rhs.partitions_;
+  };
+
+  friend std::ostream& operator<<(std::ostream& os, const ListOffsetsTopic& arg) {
+    return os << "{topic = " << arg.topic_ << ", partitions = " << arg.partitions_ << "}";
+  }
+};
+
+class ListOffsetsRequest : public Request {
+public:
+  // v0 .. v1
+  ListOffsetsRequest(INT32 replica_id, NULLABLE_ARRAY<ListOffsetsTopic> topics)
+      : ListOffsetsRequest(replica_id, -1, topics){};
+
+  // v2
+  ListOffsetsRequest(INT32 replica_id, INT8 isolation_level,
+                     NULLABLE_ARRAY<ListOffsetsTopic> topics)
+      : Request{RequestType::ListOffsets}, replica_id_{replica_id},
+        isolation_level_{isolation_level}, topics_{topics} {};
+
+  bool operator==(const ListOffsetsRequest& rhs) const {
+    return request_header_ == rhs.request_header_ && replica_id_ == rhs.replica_id_ &&
+           isolation_level_ == rhs.isolation_level_ && topics_ == rhs.topics_;
+  };
+
+protected:
+  size_t encodeDetails(char* dst, Encoder& encoder) const override {
+    size_t written{0};
+    written += encoder.encode(replica_id_, dst + written);
+    written += encoder.encode(isolation_level_, dst + written);
+    written += encoder.encode(topics_, dst + written);
+    return written;
+  }
+
+  std::ostream& printDetails(std::ostream& os) const override {
+    return os << "{replica_id = " << replica_id_
+              << ", isolation_level = " << static_cast<uint32_t>(isolation_level_)
+              << ", topics = " << topics_ << "}";
+  }
+
+private:
+  INT32 replica_id_;
+  INT8 isolation_level_;
+  NULLABLE_ARRAY<ListOffsetsTopic> topics_;
+};
+
+class ListOffsetsPartitionV0Buffer
+    : public CompositeBuffer<ListOffsetsPartition, Int32Buffer, Int64Buffer, Int32Buffer> {};
+class ListOffsetsPartitionV0ArrayBuffer
+    : public ArrayBuffer<ListOffsetsPartition, ListOffsetsPartitionV0Buffer> {};
+class ListOffsetsTopicV0Buffer
+    : public CompositeBuffer<ListOffsetsTopic, StringBuffer, ListOffsetsPartitionV0ArrayBuffer> {};
+class ListOffsetsTopicV0ArrayBuffer
+    : public ArrayBuffer<ListOffsetsTopic, ListOffsetsTopicV0Buffer> {};
+
+class ListOffsetsPartitionV1Buffer
+    : public CompositeBuffer<ListOffsetsPartition, Int32Buffer, Int64Buffer> {};
+class ListOffsetsPartitionV1ArrayBuffer
+    : public ArrayBuffer<ListOffsetsPartition, ListOffsetsPartitionV1Buffer> {};
+class ListOffsetsTopicV1Buffer
+    : public CompositeBuffer<ListOffsetsTopic, StringBuffer, ListOffsetsPartitionV1ArrayBuffer> {};
+class ListOffsetsTopicV1ArrayBuffer
+    : public ArrayBuffer<ListOffsetsTopic, ListOffsetsTopicV1Buffer> {};
+
+class ListOffsetsRequestV0Buffer
+    : public CompositeBuffer<ListOffsetsRequest, Int32Buffer, ListOffsetsTopicV0ArrayBuffer> {};
+class ListOffsetsRequestV1Buffer
+    : public CompositeBuffer<ListOffsetsRequest, Int32Buffer, ListOffsetsTopicV1ArrayBuffer> {};
+class ListOffsetsRequestV2Buffer
+    : public CompositeBuffer<ListOffsetsRequest, Int32Buffer, Int8Buffer,
+                             ListOffsetsTopicV1ArrayBuffer> {};
+
+DEFINE_REQUEST_PARSER(ListOffsetsRequest, V0);
+DEFINE_REQUEST_PARSER(ListOffsetsRequest, V1);
+DEFINE_REQUEST_PARSER(ListOffsetsRequest, V2);
+
+// === METADATA (3) ============================================================
+
+class MetadataRequest : public Request {
+public:
+  // v0 .. v3
+  MetadataRequest(NULLABLE_ARRAY<STRING> topics) : MetadataRequest(topics, false){};
+
+  // v4
+  MetadataRequest(NULLABLE_ARRAY<STRING> topics, BOOLEAN allow_auto_topic_creation)
+      : Request{RequestType::Metadata}, topics_{topics}, allow_auto_topic_creation_{
+                                                             allow_auto_topic_creation} {};
+
+  bool operator==(const MetadataRequest& rhs) const {
+    return request_header_ == rhs.request_header_ && topics_ == rhs.topics_ &&
+           allow_auto_topic_creation_ == rhs.allow_auto_topic_creation_;
+  };
+
+protected:
+  size_t encodeDetails(char* dst, Encoder& encoder) const override {
+    size_t written{0};
+    written += encoder.encode(topics_, dst + written);
+    written += encoder.encode(allow_auto_topic_creation_, dst + written);
+    return written;
+  }
+
+  std::ostream& printDetails(std::ostream& os) const override {
+    return os << "{topics = " << topics_
+              << ", allow_auto_topic_creation = " << allow_auto_topic_creation_ << "}";
+  }
+
+private:
+  NULLABLE_ARRAY<STRING> topics_;
+  BOOLEAN allow_auto_topic_creation_;
+};
+
+class MetadataRequestTopicV0Buffer : public ArrayBuffer<STRING, StringBuffer> {};
+class MetadataRequestV0Buffer
+    : public CompositeBuffer<MetadataRequest, MetadataRequestTopicV0Buffer> {};
+class MetadataRequestV4Buffer
+    : public CompositeBuffer<MetadataRequest, MetadataRequestTopicV0Buffer, BoolBuffer> {};
+
+DEFINE_REQUEST_PARSER(MetadataRequest, V0);
+DEFINE_REQUEST_PARSER(MetadataRequest, V4);
+
+// === OFFSET COMMIT (8) =======================================================
+
+struct OffsetCommitPartition {
+  INT32 partition_;
+  INT64 offset_;
+  NULLABLE_STRING metadata_;
+
+  // v0 *and* v2
+  OffsetCommitPartition(INT32 partition, INT64 offset, NULLABLE_STRING metadata)
+      : partition_{partition}, offset_{offset}, metadata_{metadata} {};
+
+  // v1
+  OffsetCommitPartition(INT32 partition, INT64 offset, INT64 /* timestamp */,
+                        NULLABLE_STRING metadata)
+      : partition_{partition}, offset_{offset}, metadata_{metadata} {};
+
+  size_t encode(char* dst, Encoder& encoder) const {
+    size_t written{0};
+    written += encoder.encode(partition_, dst + written);
+    written += encoder.encode(offset_, dst + written);
+    written += encoder.encode(metadata_, dst + written);
+    return written;
+  }
+
+  bool operator==(const OffsetCommitPartition& rhs) const {
+    return partition_ == rhs.partition_ && offset_ == rhs.offset_ && metadata_ == rhs.metadata_;
+  };
+
+  friend std::ostream& operator<<(std::ostream& os, const OffsetCommitPartition& arg) {
+    return os << "{partition = " << arg.partition_ << ", offset = " << arg.offset_
+              << ", metadata = " << arg.metadata_ << "}";
+  }
+};
+
+struct OffsetCommitTopic {
+  STRING topic_;
+  NULLABLE_ARRAY<OffsetCommitPartition> partitions_;
+
+  size_t encode(char* dst, Encoder& encoder) const {
+    size_t written{0};
+    written += encoder.encode(topic_, dst + written);
+    written += encoder.encode(partitions_, dst + written);
+    return written;
+  }
+
+  bool operator==(const OffsetCommitTopic& rhs) const {
+    return topic_ == rhs.topic_ && partitions_ == rhs.partitions_;
+  };
+
+  friend std::ostream& operator<<(std::ostream& os, const OffsetCommitTopic& arg) {
+    return os << "{topic = " << arg.topic_ << ", partitions_ = " << arg.partitions_ << "}";
+  }
+};
+
+class OffsetCommitRequest : public Request {
+public:
+  // v0
+  OffsetCommitRequest(STRING group_id, NULLABLE_ARRAY<OffsetCommitTopic> topics)
+      : OffsetCommitRequest(group_id, -1, "no-member-id", -1, topics){};
+
+  // v1
+  OffsetCommitRequest(STRING group_id, INT32 group_generation_id, STRING member_id,
+                      NULLABLE_ARRAY<OffsetCommitTopic> topics)
+      : OffsetCommitRequest(group_id, group_generation_id, member_id, -1, topics){};
+
+  // v2 .. v3
+  OffsetCommitRequest(STRING group_id, INT32 group_generation_id, STRING member_id,
+                      INT64 retention_time, NULLABLE_ARRAY<OffsetCommitTopic> topics)
+      : Request{RequestType::OffsetCommit}, group_id_{group_id},
+        group_generation_id_{group_generation_id}, member_id_{member_id},
+        retention_time_{retention_time}, topics_{topics} {};
+
+  bool operator==(const OffsetCommitRequest& rhs) const {
+    return request_header_ == rhs.request_header_ && group_id_ == rhs.group_id_ &&
+           group_generation_id_ == rhs.group_generation_id_ && member_id_ == rhs.member_id_ &&
+           retention_time_ == rhs.retention_time_ && topics_ == rhs.topics_;
+  };
+
+protected:
+  size_t encodeDetails(char* dst, Encoder& encoder) const override {
+    size_t written{0};
+    written += encoder.encode(group_id_, dst + written);
+    written += encoder.encode(group_generation_id_, dst + written);
+    written += encoder.encode(member_id_, dst + written);
+    written += encoder.encode(retention_time_, dst + written);
+    written += encoder.encode(topics_, dst + written);
+    return written;
+  }
+
+  std::ostream& printDetails(std::ostream& os) const override {
+    return os << "{group_id = " << group_id_ << ", group_generation_id = " << group_generation_id_
+              << ", member_id = " << member_id_ << ", retention_time = " << retention_time_
+              << ", topics = " << topics_ << "}";
+  }
+
+private:
+  STRING group_id_;
+  INT32 group_generation_id_;
+  STRING member_id_;
+  INT64 retention_time_;
+  NULLABLE_ARRAY<OffsetCommitTopic> topics_;
+};
+
+class OffsetCommitPartitionV0Buffer : public CompositeBuffer<OffsetCommitPartition, Int32Buffer,
+                                                             Int64Buffer, NullableStringBuffer> {};
+class OffsetCommitPartitionV0ArrayBuffer
+    : public ArrayBuffer<OffsetCommitPartition, OffsetCommitPartitionV0Buffer> {};
+class OffsetCommitTopicV0Buffer
+    : public CompositeBuffer<OffsetCommitTopic, StringBuffer, OffsetCommitPartitionV0ArrayBuffer> {
+};
+class OffsetCommitTopicV0ArrayBuffer
+    : public ArrayBuffer<OffsetCommitTopic, OffsetCommitTopicV0Buffer> {};
+
+class OffsetCommitPartitionV1Buffer
+    : public CompositeBuffer<OffsetCommitPartition, Int32Buffer, Int64Buffer, Int64Buffer,
+                             NullableStringBuffer> {};
+class OffsetCommitPartitionV1ArrayBuffer
+    : public ArrayBuffer<OffsetCommitPartition, OffsetCommitPartitionV1Buffer> {};
+class OffsetCommitTopicV1Buffer
+    : public CompositeBuffer<OffsetCommitTopic, StringBuffer, OffsetCommitPartitionV1ArrayBuffer> {
+};
+class OffsetCommitTopicV1ArrayBuffer
+    : public ArrayBuffer<OffsetCommitTopic, OffsetCommitTopicV1Buffer> {};
+
+class OffsetCommitRequestV0Buffer
+    : public CompositeBuffer<OffsetCommitRequest, StringBuffer, OffsetCommitTopicV0ArrayBuffer> {};
+class OffsetCommitRequestV1Buffer
+    : public CompositeBuffer<OffsetCommitRequest, StringBuffer, Int32Buffer, StringBuffer,
+                             OffsetCommitTopicV1ArrayBuffer> {};
+class OffsetCommitRequestV2Buffer
+    : public CompositeBuffer<OffsetCommitRequest, StringBuffer, Int32Buffer, StringBuffer,
+                             Int64Buffer, OffsetCommitTopicV0ArrayBuffer> {};
+
+DEFINE_REQUEST_PARSER(OffsetCommitRequest, V0);
+DEFINE_REQUEST_PARSER(OffsetCommitRequest, V1);
+DEFINE_REQUEST_PARSER(OffsetCommitRequest, V2);
+
+// === OFFSET FETCH (9) ========================================================
+
+struct OffsetFetchTopic {
+  STRING topic_;
+  NULLABLE_ARRAY<INT32> partitions_;
+
+  size_t encode(char* dst, Encoder& encoder) const {
+    size_t written{0};
+    written += encoder.encode(topic_, dst + written);
+    written += encoder.encode(partitions_, dst + written);
+    return written;
+  }
+
+  bool operator==(const OffsetFetchTopic& rhs) const {
+    return topic_ == rhs.topic_ && partitions_ == rhs.partitions_;
+  };
+
+  friend std::ostream& operator<<(std::ostream& os, const OffsetFetchTopic& arg) {
+    return os << "{topic = " << arg.topic_ << ", partitions = " << arg.partitions_ << "}";
+  }
+};
+
+class OffsetFetchRequest : public Request {
+public:
+  // v0 .. v2
+  OffsetFetchRequest(STRING group_id, NULLABLE_ARRAY<OffsetFetchTopic> topics)
+      : Request{RequestType::OffsetFetch}, group_id_{group_id}, topics_{topics} {};
+
+  bool operator==(const OffsetFetchRequest& rhs) const {
+    return request_header_ == rhs.request_header_ && group_id_ == rhs.group_id_ &&
+           topics_ == rhs.topics_;
+  };
+
+protected:
+  size_t encodeDetails(char* dst, Encoder& encoder) const override {
+    size_t written{0};
+    written += encoder.encode(group_id_, dst + written);
+    written += encoder.encode(topics_, dst + written);
+    return written;
+  }
+
+  std::ostream& printDetails(std::ostream& os) const override {
+    return os << "{group_id = " << group_id_ << ", topics = " << topics_ << "}";
+  }
+
+private:
+  STRING group_id_;
+  NULLABLE_ARRAY<OffsetFetchTopic> topics_;
+};
+
+class OffsetFetchPartitionV0ArrayBuffer : public ArrayBuffer<INT32, Int32Buffer> {};
+class OffsetFetchTopicV0Buffer
+    : public CompositeBuffer<OffsetFetchTopic, StringBuffer, OffsetFetchPartitionV0ArrayBuffer> {};
+class OffsetFetchTopicV0ArrayBuffer
+    : public ArrayBuffer<OffsetFetchTopic, OffsetFetchTopicV0Buffer> {};
+
+class OffsetFetchRequestV0Buffer
+    : public CompositeBuffer<OffsetFetchRequest, StringBuffer, OffsetFetchTopicV0ArrayBuffer> {};
+class OffsetFetchRequestV2Buffer
+    : public CompositeBuffer<OffsetFetchRequest, StringBuffer, OffsetFetchTopicV0ArrayBuffer> {};
+
+DEFINE_REQUEST_PARSER(OffsetFetchRequest, V0);
+DEFINE_REQUEST_PARSER(OffsetFetchRequest, V2);
+
+// === API VERSIONS (18) =======================================================
+
+class ApiVersionsRequest : public Request {
+public:
+  // v0 .. v1
+  ApiVersionsRequest() : Request{RequestType::ApiVersions} {};
+
+  bool operator==(const ApiVersionsRequest& rhs) const {
+    return request_header_ == rhs.request_header_;
+  };
+
+protected:
+  size_t encodeDetails(char*, Encoder&) const override { return 0; }
+
+  std::ostream& printDetails(std::ostream& os) const override { return os << "{}"; }
+};
+
+class ApiVersionsRequestV0Buffer : public NullBuffer<ApiVersionsRequest> {};
+
+DEFINE_REQUEST_PARSER(ApiVersionsRequest, V0);
+
+// === UNKNOWN REQUEST =========================================================
+
+class UnknownRequest : public Request {
+public:
+  UnknownRequest(const RequestHeader& request_header) : Request{request_header} {};
+
+protected:
+  // this isn't the prettiest, as we have thrown away the data
+  // XXX(adam.kotwasinski) discuss capturing the data as-is, and simply putting it back
+  //   this would add ability to forward unknown types of requests
+  size_t encodeDetails(char*, Encoder&) const override {
+    throw EnvoyException("cannot serialize unknown request");
+  }
+
+  std::ostream& printDetails(std::ostream& out) const override {
+    return out << "{unknown request}";
+  }
+};
+
+// ignores data until the end of request (contained in c)
+class SentinelConsumer : public Parser {
+public:
+  SentinelConsumer(RequestContextSharedPtr context) : context_{context} {};
+  ParseResponse parse(const char*& buffer, uint64_t& remaining) override;
+
+private:
+  const RequestContextSharedPtr context_;
+};
+
+// === REQUEST SERIALIZER ======================================================
+
+class RequestSerializer {
+public:
+  size_t encode(const Request& request, char* dst);
+};
+
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy
diff --git a/source/extensions/filters/network/kafka/kafka_response.cc b/source/extensions/filters/network/kafka/kafka_response.cc
new file mode 100644
index 0000000000000..49d902aa8325d
--- /dev/null
+++ b/source/extensions/filters/network/kafka/kafka_response.cc
@@ -0,0 +1,15 @@
+#include "extensions/filters/network/kafka/kafka_response.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+
+ParseResponse ResponseStartParser::parse(const char*&, uint64_t&) {
+  return ParseResponse::stillWaiting();
+}
+
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy
diff --git a/source/extensions/filters/network/kafka/kafka_response.h b/source/extensions/filters/network/kafka/kafka_response.h
new file mode 100644
index 0000000000000..f3804d6d504fa
--- /dev/null
+++ b/source/extensions/filters/network/kafka/kafka_response.h
@@ -0,0 +1,28 @@
+#pragma once
+
+#include "extensions/filters/network/kafka/parser.h"
+#include "extensions/filters/network/kafka/serialization.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+
+// =============================================================================
+// === RESPONSE ================================================================
+// =============================================================================
+
+struct ResponseContext {
+  INT32 remaining_request_size_{0};
+};
+
+class ResponseStartParser : public Parser {
+public:
+  ResponseStartParser(ResponseContext&){};
+  ParseResponse parse(const char*& buffer, uint64_t& remaining);
+};
+
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy
diff --git a/source/extensions/filters/network/kafka/kafka_types.h b/source/extensions/filters/network/kafka/kafka_types.h
new file mode 100644
index 0000000000000..f5c188e2dff59
--- /dev/null
+++ b/source/extensions/filters/network/kafka/kafka_types.h
@@ -0,0 +1,31 @@
+#pragma once
+
+#include <memory>
+#include <string>
+
+#include "absl/types/optional.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+
+typedef int8_t INT8;
+typedef int16_t INT16;
+typedef int32_t INT32;
+typedef int64_t INT64;
+typedef uint32_t UINT32;
+typedef bool BOOLEAN;
+
+typedef std::string STRING;
+typedef absl::optional<STRING> NULLABLE_STRING;
+
+typedef std::vector<unsigned char> BYTES;
+typedef absl::optional<BYTES> NULLABLE_BYTES;
+
+template <typename T> using NULLABLE_ARRAY = absl::optional<std::vector<T>>;
+
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy
diff --git a/source/extensions/filters/network/kafka/metrics_holder.cc b/source/extensions/filters/network/kafka/metrics_holder.cc
new file mode 100644
index 0000000000000..2d8279d9fb45d
--- /dev/null
+++ b/source/extensions/filters/network/kafka/metrics_holder.cc
@@ -0,0 +1,40 @@
+#include "extensions/filters/network/kafka/metrics_holder.h"
+
+#include "common/common/to_lower_table.h"
+
+#include "extensions/filters/network/kafka/kafka_protocol.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+
+MetricsHolder::MetricsHolder(Stats::Scope& scope)
+    : other_handler_{scope.counter("kafka.other.count"), scope.histogram("kafka.other.count")} {
+
+  const std::vector<RequestSpec>& requests = KafkaRequest::requests();
+  metrics_.reserve(requests.size());
+  ToLowerTable lt;
+  for (auto request : requests) {
+    std::string ln = request.name_;
+    lt.toLowerCase(ln);
+
+    Stats::Counter& counter = scope.counter("kafka." + ln + ".count");
+    Stats::Histogram& histogram = scope.histogram("kafka." + ln + ".duration");
+
+    metrics_.push_back({counter, histogram});
+  };
+};
+
+RequestMetrics& MetricsHolder::metric(INT16 api_key) {
+  if (api_key >= 0 && api_key < static_cast<INT16>(metrics_.size())) {
+    return metrics_[api_key];
+  } else {
+    return other_handler_;
+  }
+}
+
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy
diff --git a/source/extensions/filters/network/kafka/metrics_holder.h b/source/extensions/filters/network/kafka/metrics_holder.h
new file mode 100644
index 0000000000000..069a7d05eccd6
--- /dev/null
+++ b/source/extensions/filters/network/kafka/metrics_holder.h
@@ -0,0 +1,32 @@
+#pragma once
+
+#include <memory>
+
+#include "envoy/stats/scope.h"
+
+#include "extensions/filters/network/kafka/kafka_types.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+
+struct RequestMetrics {
+  Stats::Counter& counter;
+  Stats::Histogram& duration;
+};
+
+class MetricsHolder {
+public:
+  MetricsHolder(Stats::Scope& scope);
+  RequestMetrics& metric(INT16 api_key_);
+
+private:
+  std::vector<RequestMetrics> metrics_;
+  RequestMetrics other_handler_;
+};
+
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy
diff --git a/source/extensions/filters/network/kafka/parser.h b/source/extensions/filters/network/kafka/parser.h
new file mode 100644
index 0000000000000..dacfdd436a569
--- /dev/null
+++ b/source/extensions/filters/network/kafka/parser.h
@@ -0,0 +1,58 @@
+#pragma once
+
+#include <sstream>
+
+#include "common/common/logger.h"
+
+#include "extensions/filters/network/kafka/kafka_types.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+
+// === PARSER ==================================================================
+
+class ParseResponse;
+
+class Parser : public Logger::Loggable<Logger::Id::kafka> {
+public:
+  virtual ParseResponse parse(const char*& buffer, uint64_t& remaining) PURE;
+  virtual ~Parser(){};
+};
+
+typedef std::shared_ptr<Parser> ParserSharedPtr;
+
+class Message {
+public:
+  virtual ~Message(){};
+
+  friend std::ostream& operator<<(std::ostream& out, const Message& arg) { return arg.print(out); }
+
+protected:
+  virtual std::ostream& print(std::ostream& os) const PURE;
+};
+
+typedef std::shared_ptr<Message> MessageSharedPtr;
+
+class ParseResponse {
+public:
+  static ParseResponse stillWaiting() { return {nullptr, nullptr}; }
+  static ParseResponse nextParser(ParserSharedPtr next_parser) { return {next_parser, nullptr}; };
+  static ParseResponse parsedMessage(MessageSharedPtr message) { return {nullptr, message}; };
+
+  bool hasData() { return (next_parser_ != nullptr) || (message_ != nullptr); }
+
+private:
+  ParseResponse(ParserSharedPtr parser, MessageSharedPtr message)
+      : next_parser_{parser}, message_{message} {};
+
+public:
+  ParserSharedPtr next_parser_;
+  MessageSharedPtr message_;
+};
+
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy
diff --git a/source/extensions/filters/network/kafka/serialization.h b/source/extensions/filters/network/kafka/serialization.h
new file mode 100644
index 0000000000000..9d8b194dda8ed
--- /dev/null
+++ b/source/extensions/filters/network/kafka/serialization.h
@@ -0,0 +1,706 @@
+#pragma once
+
+#include <algorithm>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "envoy/common/exception.h"
+#include "envoy/common/pure.h"
+
+#include "common/common/byte_order.h"
+#include "common/common/fmt.h"
+
+#include "extensions/filters/network/kafka/kafka_types.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+
+// =============================================================================
+// === DESERIALIZERS ===========================================================
+// =============================================================================
+
+/**
+ * The general idea of Buffer is that it can be feed-ed data until it is ready
+ * When true == ready(), it is safe to call get()
+ * Further feed()-ing should have no effect on a buffer
+ * (should return 0 and not move buffer/remaining)
+ */
+
+// === ABSTRACT DESERIALIZER ===================================================
+
+template <typename T> class Deserializer {
+public:
+  virtual size_t feed(const char*& buffer, uint64_t& remaining) PURE;
+  virtual bool ready() const PURE;
+  virtual T get() const PURE;
+  virtual ~Deserializer(){};
+};
+
+// === INT BUFFERS =============================================================
+
+template <typename T> class IntBuffer : public Deserializer<T> {
+public:
+  IntBuffer() : written_{0}, ready_(false){};
+
+  size_t feed(const char*& buffer, uint64_t& remaining) {
+    const size_t available = std::min<size_t>(sizeof(buf_) - written_, remaining);
+    memcpy(buf_ + written_, buffer, available);
+    written_ += available;
+
+    if (written_ == sizeof(buf_)) {
+      ready_ = true;
+    }
+
+    buffer += available;
+    remaining -= available;
+
+    return available;
+  }
+
+  bool ready() const { return ready_; }
+
+protected:
+  char buf_[sizeof(T) / sizeof(char)];
+  size_t written_;
+  bool ready_;
+};
+
+class Int8Buffer : public IntBuffer<INT8> {
+public:
+  INT8 get() const {
+    INT8 result;
+    memcpy(&result, buf_, sizeof(result));
+    return result;
+  }
+};
+
+class Int16Buffer : public IntBuffer<INT16> {
+public:
+  INT16 get() const {
+    INT16 result;
+    memcpy(&result, buf_, sizeof(result));
+    return be16toh(result);
+  }
+};
+
+class Int32Buffer : public IntBuffer<INT32> {
+public:
+  INT32 get() const {
+    INT32 result;
+    memcpy(&result, buf_, sizeof(result));
+    return be32toh(result);
+  }
+};
+
+class UInt32Buffer : public IntBuffer<UINT32> {
+public:
+  UINT32 get() const {
+    UINT32 result;
+    memcpy(&result, buf_, sizeof(result));
+    return be32toh(result);
+  }
+};
+
+class Int64Buffer : public IntBuffer<INT64> {
+public:
+  INT64 get() const {
+    INT64 result;
+    memcpy(&result, buf_, sizeof(result));
+    return be64toh(result);
+  }
+};
+
+// === BOOL BUFFER =============================================================
+
+/**
+ * Represents a boolean value in a byte.
+ * Values 0 and 1 are used to represent false and true respectively.
+ * When reading a boolean value, any non-zero value is considered true.
+ */
+class BoolBuffer : public Deserializer<BOOLEAN> {
+public:
+  BoolBuffer(){};
+
+  size_t feed(const char*& buffer, uint64_t& remaining) { return buffer_.feed(buffer, remaining); }
+
+  bool ready() const { return buffer_.ready(); }
+
+  BOOLEAN get() const { return 0 != buffer_.get(); }
+
+private:
+  Int8Buffer buffer_;
+};
+
+// === STRING BUFFER ===========================================================
+
+class StringBuffer : public Deserializer<STRING> {
+public:
+  size_t feed(const char*& buffer, uint64_t& remaining) {
+    const size_t length_consumed = length_buf_.feed(buffer, remaining);
+    if (!length_buf_.ready()) {
+      // break early: we still need to fill in length buffer
+      return length_consumed;
+    }
+
+    if (!length_consumed_) {
+      required_ = length_buf_.get();
+      if (required_ >= 0) {
+        data_buf_ = std::vector<char>(required_);
+      } else {
+        throw EnvoyException(fmt::format("invalid STRING length: {}", required_));
+      }
+      length_consumed_ = true;
+    }
+
+    const size_t data_consumed = std::min<size_t>(required_, remaining);
+    const size_t written = data_buf_.size() - required_;
+    memcpy(data_buf_.data() + written, buffer, data_consumed);
+    required_ -= data_consumed;
+
+    buffer += data_consumed;
+    remaining -= data_consumed;
+
+    if (required_ == 0) {
+      ready_ = true;
+    }
+
+    return length_consumed + data_consumed;
+  }
+
+  bool ready() const { return ready_; }
+
+  STRING get() const { return std::string(data_buf_.begin(), data_buf_.end()); }
+
+private:
+  Int16Buffer length_buf_;
+  bool length_consumed_{false};
+
+  INT16 required_;
+  std::vector<char> data_buf_;
+
+  bool ready_{false};
+};
+
+class NullableStringBuffer : public Deserializer<NULLABLE_STRING> {
+public:
+  size_t feed(const char*& buffer, uint64_t& remaining) {
+    const size_t length_consumed = length_buf_.feed(buffer, remaining);
+    if (!length_buf_.ready()) {
+      // break early: we still need to fill in length buffer
+      return length_consumed;
+    }
+
+    if (!length_consumed_) {
+      required_ = length_buf_.get();
+
+      if (required_ >= 0) {
+        data_buf_ = std::vector<char>(required_);
+      }
+      if (required_ == NULL_STRING_LENGTH) {
+        ready_ = true;
+      }
+      if (required_ < NULL_STRING_LENGTH) {
+        throw EnvoyException(fmt::format("invalid NULLABLE_STRING length: {}", required_));
+      }
+
+      length_consumed_ = true;
+    }
+
+    if (ready_) {
+      return length_consumed;
+    }
+
+    const size_t data_consumed = std::min<size_t>(required_, remaining);
+    const size_t written = data_buf_.size() - required_;
+    memcpy(data_buf_.data() + written, buffer, data_consumed);
+    required_ -= data_consumed;
+
+    buffer += data_consumed;
+    remaining -= data_consumed;
+
+    if (required_ == 0) {
+      ready_ = true;
+    }
+
+    return length_consumed + data_consumed;
+  }
+
+  bool ready() const { return ready_; }
+
+  NULLABLE_STRING get() const {
+    return required_ >= 0 ? absl::make_optional(std::string(data_buf_.begin(), data_buf_.end()))
+                          : absl::nullopt;
+  }
+
+private:
+  constexpr static INT16 NULL_STRING_LENGTH{-1};
+
+  Int16Buffer length_buf_;
+  bool length_consumed_{false};
+
+  INT16 required_;
+  std::vector<char> data_buf_;
+
+  bool ready_{false};
+};
+
+// === BYTES BUFFERS ===========================================================
+
+/**
+ * Represents a raw sequence of bytes or null.
+ * For non-null values, first the length N is given as an INT32. Then N bytes follow.
+ * A null value is encoded with length of -1 and there are no following bytes.
+ */
+
+/**
+ * This buffer ignores the data fed, the only result is the number of bytes ignored
+ */
+class NullableBytesIgnoringBuffer : public Deserializer<INT32> {
+public:
+  size_t feed(const char*& buffer, uint64_t& remaining) {
+    const size_t length_consumed = length_buf_.feed(buffer, remaining);
+    if (!length_buf_.ready()) {
+      // break early: we still need to fill in length buffer
+      return length_consumed;
+    }
+
+    if (!length_consumed_) {
+      required_max_ = length_buf_.get();
+      required_ = length_buf_.get();
+
+      if (required_ == NULL_BYTES_LENGTH) {
+        ready_ = true;
+      }
+      if (required_ < NULL_BYTES_LENGTH) {
+        throw EnvoyException(fmt::format("invalid NULLABLE_BYTES length: {}", required_));
+      }
+
+      length_consumed_ = true;
+    }
+
+    if (ready_) {
+      return length_consumed;
+    }
+
+    const size_t data_consumed = std::min<size_t>(required_, remaining);
+    required_ -= data_consumed;
+
+    buffer += data_consumed;
+    remaining -= data_consumed;
+
+    if (required_ == 0) {
+      ready_ = true;
+    }
+
+    return length_consumed + data_consumed;
+  }
+
+  bool ready() const { return ready_; }
+
+  /**
+   * Returns length of ignored array, or -1 if that was null
+   */
+  INT32 get() const { return required_max_; }
+
+private:
+  constexpr static INT32 NULL_BYTES_LENGTH{-1};
+
+  Int32Buffer length_buf_;
+  bool length_consumed_{false};
+  INT32 required_max_;
+  INT32 required_;
+  bool ready_{false};
+};
+
+/**
+ * This buffer captures the data fed
+ */
+class NullableBytesCapturingBuffer : public Deserializer<NULLABLE_BYTES> {
+public:
+  size_t feed(const char*& buffer, uint64_t& remaining) {
+    const size_t length_consumed = length_buf_.feed(buffer, remaining);
+    if (!length_buf_.ready()) {
+      // break early: we still need to fill in length buffer
+      return length_consumed;
+    }
+
+    if (!length_consumed_) {
+      required_ = length_buf_.get();
+
+      if (required_ >= 0) {
+        data_buf_ = std::vector<unsigned char>(required_);
+      }
+      if (required_ == NULL_BYTES_LENGTH) {
+        ready_ = true;
+      }
+      if (required_ < NULL_BYTES_LENGTH) {
+        throw EnvoyException(fmt::format("invalid NULLABLE_BYTES length: {}", required_));
+      }
+
+      length_consumed_ = true;
+    }
+
+    if (ready_) {
+      return length_consumed;
+    }
+
+    const size_t data_consumed = std::min<size_t>(required_, remaining);
+    const size_t written = data_buf_.size() - required_;
+    memcpy(data_buf_.data() + written, buffer, data_consumed);
+    required_ -= data_consumed;
+
+    buffer += data_consumed;
+    remaining -= data_consumed;
+
+    if (required_ == 0) {
+      ready_ = true;
+    }
+
+    return length_consumed + data_consumed;
+  }
+
+  bool ready() const { return ready_; }
+
+  NULLABLE_BYTES get() const {
+    if (NULL_BYTES_LENGTH == required_) {
+      return absl::nullopt;
+    } else {
+      return {data_buf_};
+    }
+  }
+
+private:
+  constexpr static INT32 NULL_BYTES_LENGTH{-1};
+
+  Int32Buffer length_buf_;
+  bool length_consumed_{false};
+  INT32 required_;
+
+  std::vector<unsigned char> data_buf_;
+  bool ready_{false};
+};
+
+// === COMPOSITE BUFFER ========================================================
+
+template <typename RT, typename...> class CompositeBuffer;
+
+template <typename RT, typename T1> class CompositeBuffer<RT, T1> : public Deserializer<RT> {
+public:
+  CompositeBuffer(){};
+  size_t feed(const char*& buffer, uint64_t& remaining) {
+    size_t consumed = 0;
+    consumed += buffer1_.feed(buffer, remaining);
+    return consumed;
+  }
+  bool ready() const { return buffer1_.ready(); }
+  RT get() const { return {buffer1_.get()}; }
+
+protected:
+  T1 buffer1_;
+};
+
+template <typename RT, typename T1, typename T2>
+class CompositeBuffer<RT, T1, T2> : public Deserializer<RT> {
+public:
+  CompositeBuffer(){};
+  size_t feed(const char*& buffer, uint64_t& remaining) {
+    size_t consumed = 0;
+    consumed += buffer1_.feed(buffer, remaining);
+    consumed += buffer2_.feed(buffer, remaining);
+    return consumed;
+  }
+  bool ready() const { return buffer2_.ready(); }
+  RT get() const { return {buffer1_.get(), buffer2_.get()}; }
+
+protected:
+  T1 buffer1_;
+  T2 buffer2_;
+};
+
+template <typename RT, typename T1, typename T2, typename T3>
+class CompositeBuffer<RT, T1, T2, T3> : public Deserializer<RT> {
+public:
+  CompositeBuffer(){};
+  size_t feed(const char*& buffer, uint64_t& remaining) {
+    size_t consumed = 0;
+    consumed += buffer1_.feed(buffer, remaining);
+    consumed += buffer2_.feed(buffer, remaining);
+    consumed += buffer3_.feed(buffer, remaining);
+    return consumed;
+  }
+  bool ready() const { return buffer3_.ready(); }
+  RT get() const { return {buffer1_.get(), buffer2_.get(), buffer3_.get()}; }
+
+protected:
+  T1 buffer1_;
+  T2 buffer2_;
+  T3 buffer3_;
+};
+
+template <typename RT, typename T1, typename T2, typename T3, typename T4>
+class CompositeBuffer<RT, T1, T2, T3, T4> : public Deserializer<RT> {
+public:
+  CompositeBuffer(){};
+  size_t feed(const char*& buffer, uint64_t& remaining) {
+    size_t consumed = 0;
+    consumed += buffer1_.feed(buffer, remaining);
+    consumed += buffer2_.feed(buffer, remaining);
+    consumed += buffer3_.feed(buffer, remaining);
+    consumed += buffer4_.feed(buffer, remaining);
+    return consumed;
+  }
+  bool ready() const { return buffer4_.ready(); }
+  RT get() const { return {buffer1_.get(), buffer2_.get(), buffer3_.get(), buffer4_.get()}; }
+
+protected:
+  T1 buffer1_;
+  T2 buffer2_;
+  T3 buffer3_;
+  T4 buffer4_;
+};
+
+template <typename RT, typename T1, typename T2, typename T3, typename T4, typename T5>
+class CompositeBuffer<RT, T1, T2, T3, T4, T5> : public Deserializer<RT> {
+public:
+  CompositeBuffer(){};
+  size_t feed(const char*& buffer, uint64_t& remaining) {
+    size_t consumed = 0;
+    consumed += buffer1_.feed(buffer, remaining);
+    consumed += buffer2_.feed(buffer, remaining);
+    consumed += buffer3_.feed(buffer, remaining);
+    consumed += buffer4_.feed(buffer, remaining);
+    consumed += buffer5_.feed(buffer, remaining);
+    return consumed;
+  }
+  bool ready() const { return buffer5_.ready(); }
+  RT get() const {
+    return {buffer1_.get(), buffer2_.get(), buffer3_.get(), buffer4_.get(), buffer5_.get()};
+  }
+
+protected:
+  T1 buffer1_;
+  T2 buffer2_;
+  T3 buffer3_;
+  T4 buffer4_;
+  T5 buffer5_;
+};
+
+template <typename RT, typename T1, typename T2, typename T3, typename T4, typename T5, typename T6>
+class CompositeBuffer<RT, T1, T2, T3, T4, T5, T6> : public Deserializer<RT> {
+public:
+  CompositeBuffer(){};
+  size_t feed(const char*& buffer, uint64_t& remaining) {
+    size_t consumed = 0;
+    consumed += buffer1_.feed(buffer, remaining);
+    consumed += buffer2_.feed(buffer, remaining);
+    consumed += buffer3_.feed(buffer, remaining);
+    consumed += buffer4_.feed(buffer, remaining);
+    consumed += buffer5_.feed(buffer, remaining);
+    consumed += buffer6_.feed(buffer, remaining);
+    return consumed;
+  }
+  bool ready() const { return buffer6_.ready(); }
+  RT get() const {
+    return {buffer1_.get(), buffer2_.get(), buffer3_.get(),
+            buffer4_.get(), buffer5_.get(), buffer6_.get()};
+  }
+
+protected:
+  T1 buffer1_;
+  T2 buffer2_;
+  T3 buffer3_;
+  T4 buffer4_;
+  T5 buffer5_;
+  T6 buffer6_;
+};
+
+// === ARRAY BUFFER ============================================================
+
+/**
+ * Represents a sequence of objects of a given type T. Type T can be either a primitive type (e.g.
+ * STRING) or a structure. First, the length N is given as an INT32. Then N instances of type T
+ * follow. A null array is represented with a length of -1.
+ */
+
+template <typename RT, typename CT> class ArrayBuffer : public Deserializer<NULLABLE_ARRAY<RT>> {
+public:
+  size_t feed(const char*& buffer, uint64_t& remaining) {
+
+    const size_t length_consumed = length_buf_.feed(buffer, remaining);
+    if (!length_buf_.ready()) {
+      // break early: we still need to fill in length buffer
+      return length_consumed;
+    }
+
+    if (!length_consumed_) {
+      required_ = length_buf_.get();
+
+      if (required_ >= 0) {
+        children_ = std::vector<CT>(required_);
+      }
+      if (required_ == NULL_ARRAY_LENGTH) {
+        ready_ = true;
+      }
+      if (required_ < NULL_ARRAY_LENGTH) {
+        throw EnvoyException(fmt::format("invalid array length: {}", required_));
+      }
+
+      length_consumed_ = true;
+    }
+
+    if (ready_) {
+      return length_consumed;
+    }
+
+    size_t child_consumed{0};
+    for (CT& child : children_) {
+      child_consumed += child.feed(buffer, remaining);
+    }
+
+    bool children_ready_ = true;
+    for (CT& child : children_) {
+      children_ready_ &= child.ready();
+    }
+    ready_ = children_ready_;
+
+    return length_consumed + child_consumed;
+  }
+
+  bool ready() const { return ready_; }
+
+  NULLABLE_ARRAY<RT> get() const {
+    if (NULL_ARRAY_LENGTH != required_) {
+      std::vector<RT> result{};
+      result.reserve(children_.size());
+      for (const CT& child : children_) {
+        const RT child_result = child.get();
+        result.push_back(child_result);
+      }
+      return {result};
+    } else {
+      return absl::nullopt;
+    }
+  }
+
+private:
+  constexpr static INT32 NULL_ARRAY_LENGTH{-1};
+
+  Int32Buffer length_buf_;
+  bool length_consumed_{false};
+  INT32 required_;
+  std::vector<CT> children_;
+  bool children_setup_{false};
+  bool ready_{false};
+};
+
+// === NULL BUFFER =============================================================
+
+template <typename RT> class NullBuffer : public Deserializer<RT> {
+public:
+  size_t feed(const char*&, uint64_t&) { return 0; }
+
+  bool ready() const { return true; }
+
+  RT get() const { return {}; }
+};
+
+// =============================================================================
+// === SERIALIZER ==============================================================
+// =============================================================================
+
+/**
+ * Encoder serializes provided argument to Kafka format
+ * In case of primitive types, this is done explicitly
+ * In case of composite types (request/response), this is done by calling 'encode' on provided
+ * argument
+ */
+
+class Encoder {
+public:
+  template <typename T> size_t encode(const T& arg, char* dst);
+
+  template <typename T> size_t encode(const NULLABLE_ARRAY<T>& arg, char* dst);
+};
+
+template <typename T> inline size_t Encoder::encode(const T& arg, char* dst) {
+  return arg.encode(dst, *this);
+}
+
+template <> inline size_t Encoder::encode(const INT8& arg, char* dst) {
+  memcpy(dst, &arg, sizeof(INT8));
+  return sizeof(INT8);
+}
+
+#define ENCODE_NUMERIC_TYPE(TYPE, CONVERTER)                                                       \
+  template <> inline size_t Encoder::encode(const TYPE& arg, char* dst) {                          \
+    TYPE val = CONVERTER(arg);                                                                     \
+    memcpy(dst, &val, sizeof(TYPE));                                                               \
+    return sizeof(TYPE);                                                                           \
+  }
+
+ENCODE_NUMERIC_TYPE(INT16, htobe16);
+ENCODE_NUMERIC_TYPE(INT32, htobe32);
+ENCODE_NUMERIC_TYPE(UINT32, htobe32);
+ENCODE_NUMERIC_TYPE(INT64, htobe64);
+
+template <> inline size_t Encoder::encode(const BOOLEAN& arg, char* dst) {
+  INT8 val = arg;
+  memcpy(dst, &val, sizeof(INT8));
+  return sizeof(INT8);
+}
+
+template <> inline size_t Encoder::encode(const STRING& arg, char* dst) {
+  INT16 string_length = arg.length();
+  size_t header_length = encode(string_length, dst);
+  memcpy(dst + header_length, arg.c_str(), string_length);
+  return header_length + string_length;
+}
+
+template <> inline size_t Encoder::encode(const NULLABLE_STRING& arg, char* dst) {
+  if (arg.has_value()) {
+    return encode(*arg, dst);
+  } else {
+    INT16 len = -1;
+    return encode(len, dst);
+  }
+}
+
+template <> inline size_t Encoder::encode(const BYTES& arg, char* dst) {
+  INT32 data_length = arg.size();
+  size_t header_length = encode(data_length, dst);
+  memcpy(dst + header_length, arg.data(), arg.size());
+  return header_length + data_length;
+}
+
+template <> inline size_t Encoder::encode(const NULLABLE_BYTES& arg, char* dst) {
+  if (arg.has_value()) {
+    return encode(*arg, dst);
+  } else {
+    INT32 len = -1;
+    return encode(len, dst);
+  }
+}
+
+// encoding array
+template <typename T> size_t Encoder::encode(const NULLABLE_ARRAY<T>& arg, char* dst) {
+  if (arg.has_value()) {
+    INT32 len = arg->size();
+    size_t header_length = encode(len, dst);
+    size_t written{0};
+    for (const T& el : *arg) {
+      written += encode(el, dst + header_length + written);
+    }
+    return header_length + written;
+  } else {
+    INT32 len = -1;
+    return encode(len, dst);
+  }
+}
+
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy
diff --git a/source/extensions/filters/network/well_known_names.h b/source/extensions/filters/network/well_known_names.h
index de09145f83a3c..97984f9a436b8 100644
--- a/source/extensions/filters/network/well_known_names.h
+++ b/source/extensions/filters/network/well_known_names.h
@@ -28,6 +28,8 @@ class NetworkFilterNameValues {
   const std::string TcpProxy = "envoy.tcp_proxy";
   // Authorization filter
   const std::string ExtAuthorization = "envoy.ext_authz";
+  // Kafka filter
+  const std::string Kafka = "envoy.filters.network.kafka";
   // Thrift proxy filter
   const std::string ThriftProxy = "envoy.filters.network.thrift_proxy";
   // Role based access control filter
diff --git a/test/extensions/filters/network/kafka/BUILD b/test/extensions/filters/network/kafka/BUILD
new file mode 100644
index 0000000000000..9aa4d56ec69d7
--- /dev/null
+++ b/test/extensions/filters/network/kafka/BUILD
@@ -0,0 +1,32 @@
+licenses(["notice"])  # Apache 2
+
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_package",
+)
+load(
+    "//test/extensions:extensions_build_system.bzl",
+    "envoy_extension_cc_test",
+)
+
+envoy_package()
+
+envoy_extension_cc_test(
+    name = "serialization_test",
+    srcs = ["serialization_test.cc"],
+    extension_name = "envoy.filters.network.kafka",
+    deps = [
+        "//source/extensions/filters/network/kafka:serialization_lib",
+        "//test/mocks/server:server_mocks",
+    ],
+)
+
+envoy_extension_cc_test(
+    name = "codec_test",
+    srcs = ["codec_test.cc"],
+    extension_name = "envoy.filters.network.kafka",
+    deps = [
+        "//source/extensions/filters/network/kafka:kafka_codec_lib",
+        "//test/mocks/server:server_mocks",
+    ],
+)
diff --git a/test/extensions/filters/network/kafka/codec_test.cc b/test/extensions/filters/network/kafka/codec_test.cc
new file mode 100644
index 0000000000000..f570c0b66eba1
--- /dev/null
+++ b/test/extensions/filters/network/kafka/codec_test.cc
@@ -0,0 +1,199 @@
+#include "extensions/filters/network/kafka/codec.h"
+
+#include "test/mocks/server/mocks.h"
+
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+using testing::_;
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+
+class RequestDecoderTest : public testing::Test {
+public:
+  Buffer::OwnedImpl buffer_;
+
+  template <typename T> std::shared_ptr<T> serializeAndDeserialize(T request);
+};
+
+class MockMessageListener : public MessageListener {
+public:
+  MOCK_METHOD1(onMessage, void(MessageSharedPtr));
+};
+
+template <typename T> std::shared_ptr<T> RequestDecoderTest::serializeAndDeserialize(T request) {
+  char data[1024];
+  RequestSerializer serializer;
+  size_t written = serializer.encode(request, data);
+  buffer_.add(data, written);
+
+  std::shared_ptr<MockMessageListener> mock_listener = std::make_shared<MockMessageListener>();
+  RequestDecoder testee{RequestParserResolver::KAFKA_0_11, {mock_listener}};
+
+  MessageSharedPtr receivedMessage;
+  EXPECT_CALL(*mock_listener, onMessage(_)).WillOnce(testing::SaveArg<0>(&receivedMessage));
+
+  testee.onData(buffer_);
+
+  return std::dynamic_pointer_cast<T>(receivedMessage);
+};
+
+// === FETCH (1) ============================================================
+
+TEST_F(RequestDecoderTest, shouldParseFetchRequest) {
+  // given
+  FetchRequest request{1,
+                       1000,
+                       10,
+                       20,
+                       2,
+                       {{
+                           {"topic1", {{{10, 20, 1000, 2000}}}},
+                           {"topic1", {{{11, 21, 1001, 2001}, {12, 22, 1002, 2002}}}},
+                           {"topic1", {{{13, 23, 1003, 2003}}}},
+                       }}};
+  request.apiVersion() = 5;
+  request.correlationId() = 10;
+  request.clientId() = "client-id";
+
+  // when
+  auto received = serializeAndDeserialize(request);
+
+  // then
+  ASSERT_NE(received, nullptr);
+  ASSERT_EQ(*received, request);
+}
+
+// === LIST OFFSETS (2) ========================================================
+
+TEST_F(RequestDecoderTest, shouldParseListOffsetsRequest) {
+  // given
+  ListOffsetsRequest request{10,
+                             2,
+                             {{
+                                 {"topic1", {{{1, 1000}, {2, 2000}}}},
+                                 {"topic2", {{{3, 3000}}}},
+                             }}};
+  request.apiVersion() = 2;
+  request.correlationId() = 10;
+  request.clientId() = "client-id";
+
+  // when
+  auto received = serializeAndDeserialize(request);
+
+  // then
+  ASSERT_NE(received, nullptr);
+  ASSERT_EQ(*received, request);
+}
+
+// === METADATA (3) ============================================================
+
+TEST_F(RequestDecoderTest, shouldParseMetadataRequest) {
+  // given
+  MetadataRequest request{{{"t1", "t2", "t3"}}, true};
+  request.apiVersion() = 4;
+  request.correlationId() = 10;
+  request.clientId() = "client-id";
+
+  // when
+  auto received = serializeAndDeserialize(request);
+
+  // then
+  ASSERT_NE(received, nullptr);
+  ASSERT_EQ(*received, request);
+}
+
+// === OFFSET COMMIT (8) =======================================================
+
+TEST_F(RequestDecoderTest, shouldParseOffsetCommitRequest) {
+  // given
+  OffsetCommitRequest request{
+      "group_id",
+      1234,
+      "member",
+      2345,
+      {{{"topic1", {{{0, 10, "m1"}, {2, 20, "m2"}}}}, {"topic2", {{{3, 30, "m3"}}}}}}};
+  request.apiVersion() = 3;
+  request.correlationId() = 10;
+  request.clientId() = "client-id";
+
+  // when
+  auto received = serializeAndDeserialize(request);
+
+  // then
+  ASSERT_NE(received, nullptr);
+  ASSERT_EQ(*received, request);
+}
+
+// === OFFSET FETCH (9) ========================================================
+
+TEST_F(RequestDecoderTest, shouldParseOffsetFetchRequest) {
+  // given
+  OffsetFetchRequest request{"group_id", {{{"topic1", {{0, 1, 2}}}, {"topic2", {{3, 4}}}}}};
+
+  request.apiVersion() = 3;
+  request.correlationId() = 10;
+  request.clientId() = "client-id";
+
+  // when
+  auto received = serializeAndDeserialize(request);
+
+  // then
+  ASSERT_NE(received, nullptr);
+  ASSERT_EQ(*received, request);
+}
+
+// === API VERSIONS (18) =======================================================
+
+TEST_F(RequestDecoderTest, shouldParseApiVersionsRequest) {
+  // given
+  ApiVersionsRequest request{};
+  request.apiVersion() = 1;
+  request.correlationId() = 42;
+  request.clientId() = "client-id";
+
+  // when
+  auto received = serializeAndDeserialize(request);
+
+  // then
+  ASSERT_NE(received, nullptr);
+  ASSERT_EQ(*received, request);
+}
+
+// === UNKNOWN REQUEST =========================================================
+
+TEST_F(RequestDecoderTest, shouldProduceAbortedMessageOnUnknownData) {
+  // given
+  RequestSerializer serializer;
+  ApiVersionsRequest request{};
+  request.apiVersion() = 1;
+  request.correlationId() = 42;
+  request.clientId() = "client-id";
+
+  char data[1024];
+  size_t written = serializer.encode(request, data);
+
+  buffer_.add(data, written);
+
+  std::shared_ptr<MockMessageListener> mock_listener = std::make_shared<MockMessageListener>();
+  RequestParserResolver parser_resolver{{}}; // we do not accept any kind of message here
+  RequestDecoder testee{parser_resolver, {mock_listener}};
+
+  MessageSharedPtr rev;
+  EXPECT_CALL(*mock_listener, onMessage(_)).WillOnce(testing::SaveArg<0>(&rev));
+
+  // when
+  testee.onData(buffer_);
+
+  // then
+  auto received = std::dynamic_pointer_cast<UnknownRequest>(rev);
+  ASSERT_NE(received, nullptr);
+}
+
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy
diff --git a/test/extensions/filters/network/kafka/serialization_test.cc b/test/extensions/filters/network/kafka/serialization_test.cc
new file mode 100644
index 0000000000000..8859e8932d090
--- /dev/null
+++ b/test/extensions/filters/network/kafka/serialization_test.cc
@@ -0,0 +1,400 @@
+#include "extensions/filters/network/kafka/serialization.h"
+
+#include "test/mocks/server/mocks.h"
+
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+using testing::_;
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+
+// === EMPTY (FRESHLY INITIALIZED) BUFFER TESTS ================================
+
+// freshly created buffers should not be ready
+#define TEST_EmptyBufferShouldNotBeReady(BufferClass)                                              \
+  TEST(BufferClass, EmptyBufferShouldNotBeReady) {                                                 \
+    const BufferClass testee{};                                                                    \
+    ASSERT_EQ(testee.ready(), false);                                                              \
+  }
+
+TEST_EmptyBufferShouldNotBeReady(Int8Buffer);
+TEST_EmptyBufferShouldNotBeReady(Int16Buffer);
+TEST_EmptyBufferShouldNotBeReady(Int32Buffer);
+TEST_EmptyBufferShouldNotBeReady(UInt32Buffer);
+TEST_EmptyBufferShouldNotBeReady(Int64Buffer);
+TEST_EmptyBufferShouldNotBeReady(BoolBuffer);
+TEST_EmptyBufferShouldNotBeReady(StringBuffer);
+TEST_EmptyBufferShouldNotBeReady(NullableStringBuffer);
+TEST_EmptyBufferShouldNotBeReady(NullableBytesIgnoringBuffer);
+TEST(CompositeBuffer, EmptyBufferShouldNotBeReady) {
+  // given
+  const CompositeBuffer<INT8, Int8Buffer> testee{};
+  // when, then
+  ASSERT_EQ(testee.ready(), false);
+}
+TEST(ArrayBuffer, EmptyBufferShouldNotBeReady) {
+  // given
+  const ArrayBuffer<INT8, Int8Buffer> testee{};
+  // when, then
+  ASSERT_EQ(testee.ready(), false);
+}
+
+// Null buffer is a special case, it's always ready and can provide results via 0-arg ctor
+TEST(NullBuffer, EmptyBufferShouldBeReady) {
+  // given
+  const NullBuffer<INT8> testee{};
+  // when, then
+  ASSERT_EQ(testee.ready(), true);
+  ASSERT_EQ(testee.get(), 0);
+}
+
+// === SERIALIZATION / DESERIALIZATION TESTS ===================================
+
+Encoder encoder;
+
+// exactly what is says on the tin:
+// 1. serialize expected using Encoder
+// 2. deserialize byte array using testee buffer
+// 3. verify result = expected
+// 4. verify that data pointer moved correct amount
+// 5. feed testee more data
+// 6. verify that nothing more was consumed
+template <typename BT, typename AT>
+void serializeThenDeserializeAndCheckEqualityInOneGo(AT expected) {
+  // given
+  BT testee{};
+
+  uint64_t remaining = 1024;
+  const uint64_t orig_remaining = remaining;
+  const std::unique_ptr<char[]> holder = std::unique_ptr<char[]>(new char[remaining]);
+
+  const size_t written = encoder.encode(expected, holder.get());
+
+  const char* data = holder.get();
+  const char* orig_data = data;
+
+  // when
+  const size_t consumed = testee.feed(data, remaining);
+
+  // then
+  ASSERT_EQ(consumed, written);
+  ASSERT_EQ(testee.ready(), true);
+  ASSERT_EQ(testee.get(), expected);
+  ASSERT_EQ(data, orig_data + consumed);
+  ASSERT_EQ(remaining, orig_remaining - consumed);
+
+  // when - 2
+  const size_t consumed2 = testee.feed(data, remaining);
+
+  // then - 2 (nothing changes)
+  ASSERT_EQ(consumed2, 0);
+  ASSERT_EQ(data, orig_data + consumed);
+  ASSERT_EQ(remaining, orig_remaining - consumed);
+}
+
+// does the same thing as the above test,
+// but instead of providing whole data at one, it provides it in N one-byte chunks
+// this verifies if buffer keeps state properly
+template <typename BT, typename AT>
+void serializeThenDeserializeAndCheckEqualityWithChunks(AT expected) {
+  // given
+  BT testee{};
+
+  const std::unique_ptr<char[]> holder = std::unique_ptr<char[]>(new char[1024]);
+
+  const size_t written = encoder.encode(expected, holder.get());
+
+  const char* data = holder.get();
+  const char* orig_data = data;
+
+  // when
+  size_t consumed = 0;
+  for (size_t i = 0; i < written; ++i) {
+    uint64_t data_size = 1;
+    consumed += testee.feed(data, data_size);
+    ASSERT_EQ(data_size, 0);
+  }
+
+  // then
+  ASSERT_EQ(consumed, written);
+  ASSERT_EQ(testee.ready(), true);
+  ASSERT_EQ(testee.get(), expected);
+  ASSERT_EQ(data, orig_data + consumed);
+
+  // when - 2
+  uint64_t remaining = 1024;
+  const size_t consumed2 = testee.feed(data, remaining);
+
+  // then - 2 (nothing changes)
+  ASSERT_EQ(consumed2, 0);
+  ASSERT_EQ(data, orig_data + consumed);
+  ASSERT_EQ(remaining, 1024);
+}
+
+template <typename BT, typename AT> void serializeThenDeserializeAndCheckEquality(AT expected) {
+  serializeThenDeserializeAndCheckEqualityInOneGo<BT>(expected);
+  serializeThenDeserializeAndCheckEqualityWithChunks<BT>(expected);
+}
+
+// === NUMERIC BUFFERS =========================================================
+
+// macroed out test for numeric buffers
+#define TEST_BufferShouldDeserialize(BufferClass, DataClass, Value)                                \
+  TEST(DataClass, ShouldConsumeCorrectAmountOfData) {                                              \
+    /* given */                                                                                    \
+    const DataClass value = Value;                                                                 \
+    serializeThenDeserializeAndCheckEquality<BufferClass>(value);                                  \
+  }
+
+TEST_BufferShouldDeserialize(Int8Buffer, INT8, 42);
+TEST_BufferShouldDeserialize(Int16Buffer, INT16, 42);
+TEST_BufferShouldDeserialize(Int32Buffer, INT32, 42);
+TEST_BufferShouldDeserialize(UInt32Buffer, UINT32, 42);
+TEST_BufferShouldDeserialize(Int64Buffer, INT64, 42);
+TEST_BufferShouldDeserialize(BoolBuffer, BOOLEAN, true);
+
+// === (NULLABLE) STRING BUFFER ================================================
+
+TEST(StringBuffer, ShouldDeserialize) {
+  const STRING value = "sometext";
+  serializeThenDeserializeAndCheckEquality<StringBuffer>(value);
+}
+
+TEST(StringBuffer, ShouldDeserializeEmptyString) {
+  const STRING value = "";
+  serializeThenDeserializeAndCheckEquality<StringBuffer>(value);
+}
+
+TEST(StringBuffer, ShouldThrowOnInvalidLength) {
+  // given
+  StringBuffer testee;
+
+  uint64_t remaining = 1024;
+  const std::unique_ptr<char[]> holder = std::unique_ptr<char[]>(new char[remaining]);
+
+  INT16 len = -1;
+  encoder.encode(len, holder.get());
+
+  const char* data = holder.get();
+
+  // when
+  // then
+  EXPECT_THROW(testee.feed(data, remaining), EnvoyException);
+}
+
+TEST(NullableStringBuffer, ShouldDeserializeString) {
+  // given
+  const NULLABLE_STRING value{"sometext"};
+  serializeThenDeserializeAndCheckEquality<NullableStringBuffer>(value);
+}
+
+TEST(NullableStringBuffer, ShouldDeserializeEmptyString) {
+  // given
+  const NULLABLE_STRING value{""};
+  serializeThenDeserializeAndCheckEquality<NullableStringBuffer>(value);
+}
+
+TEST(NullableStringBuffer, ShouldDeserializeAbsentString) {
+  // given
+  const NULLABLE_STRING value = absl::nullopt;
+  serializeThenDeserializeAndCheckEquality<NullableStringBuffer>(value);
+}
+
+TEST(NullableStringBuffer, ShouldThrowOnInvalidLength) {
+  // given
+  NullableStringBuffer testee;
+
+  uint64_t remaining = 1024;
+  const std::unique_ptr<char[]> holder = std::unique_ptr<char[]>(new char[remaining]);
+
+  INT16 len = -2; // -1 is OK for NULLABLE_STRING
+  encoder.encode(len, holder.get());
+
+  const char* data = holder.get();
+
+  // when
+  // then
+  EXPECT_THROW(testee.feed(data, remaining), EnvoyException);
+}
+
+// === NULLABLE BYTES IGNORING BUFFER ==========================================
+
+TEST(NullableBytesIgnoringBuffer, ShouldDeserialize) {
+  // given
+  NullableBytesIgnoringBuffer testee;
+
+  uint64_t remaining = 1024;
+  const uint64_t orig_remaining = remaining;
+  const std::unique_ptr<char[]> holder = std::unique_ptr<char[]>(new char[remaining]);
+
+  const INT32 bytes_to_ignore = 100;
+  const size_t header_size = encoder.encode(bytes_to_ignore, holder.get());
+
+  const char* data = holder.get();
+  const char* orig_data = data;
+
+  // when
+  const size_t consumed = testee.feed(data, remaining);
+
+  // then
+  ASSERT_EQ(consumed, header_size + bytes_to_ignore);
+  ASSERT_EQ(testee.ready(), true);
+  ASSERT_EQ(testee.get(), bytes_to_ignore);
+  ASSERT_EQ(data, orig_data + consumed);
+  ASSERT_EQ(remaining, orig_remaining - consumed);
+
+  // when - 2
+  const size_t consumed2 = testee.feed(data, remaining);
+
+  // then - 2 (nothing changes)
+  ASSERT_EQ(consumed2, 0);
+  ASSERT_EQ(data, orig_data + consumed);
+  ASSERT_EQ(remaining, orig_remaining - consumed);
+}
+
+TEST(NullableBytesIgnoringBuffer, ShouldDeserializeNullBytes) {
+  // given
+  NullableBytesIgnoringBuffer testee;
+
+  uint64_t remaining = 1024;
+  const uint64_t orig_remaining = remaining;
+  const std::unique_ptr<char[]> holder = std::unique_ptr<char[]>(new char[remaining]);
+
+  const INT32 bytes_length = -1;
+  const size_t header_size = encoder.encode(bytes_length, holder.get());
+
+  const char* data = holder.get();
+  const char* orig_data = data;
+
+  // when
+  const size_t consumed = testee.feed(data, remaining);
+
+  // then
+  ASSERT_EQ(consumed, header_size);
+  ASSERT_EQ(testee.ready(), true);
+  ASSERT_EQ(testee.get(), bytes_length);
+  ASSERT_EQ(data, orig_data + consumed);
+  ASSERT_EQ(remaining, orig_remaining - consumed);
+
+  // when - 2
+  const size_t consumed2 = testee.feed(data, remaining);
+
+  // then - 2 (nothing changes)
+  ASSERT_EQ(consumed2, 0);
+  ASSERT_EQ(data, orig_data + consumed);
+  ASSERT_EQ(remaining, orig_remaining - consumed);
+}
+
+TEST(NullableBytesIgnoringBuffer, ShouldThrowOnInvalidLength) {
+  // given
+  NullableBytesIgnoringBuffer testee;
+
+  uint64_t remaining = 1024;
+  const std::unique_ptr<char[]> holder = std::unique_ptr<char[]>(new char[remaining]);
+
+  INT32 len = -2; // -1 is OK for NULLABLE_BYTES
+  encoder.encode(len, holder.get());
+
+  const char* data = holder.get();
+
+  // when
+  // then
+  EXPECT_THROW(testee.feed(data, remaining), EnvoyException);
+}
+
+// === NULLABLE BYTES CAPTURING BUFFER =========================================
+
+TEST(NullableBytesCapturingBuffer, ShouldDeserialize) {
+  const NULLABLE_BYTES value{{'a', 'b', 'c', 'd'}};
+  serializeThenDeserializeAndCheckEquality<NullableBytesCapturingBuffer>(value);
+}
+
+TEST(NullableBytesCapturingBuffer, ShouldDeserializeEmptyBytes) {
+  const NULLABLE_BYTES value{{}};
+  serializeThenDeserializeAndCheckEquality<NullableBytesCapturingBuffer>(value);
+}
+
+TEST(NullableBytesCapturingBuffer, ShouldDeserializeNullBytes) {
+  const NULLABLE_BYTES value = absl::nullopt;
+  serializeThenDeserializeAndCheckEquality<NullableBytesCapturingBuffer>(value);
+}
+
+TEST(NullableBytesCapturingBuffer, ShouldThrowOnInvalidLength) {
+  // given
+  NullableBytesCapturingBuffer testee;
+
+  uint64_t remaining = 1024;
+  const std::unique_ptr<char[]> holder = std::unique_ptr<char[]>(new char[remaining]);
+
+  INT16 len = -2; // -1 is OK for NULLABLE_BYTES
+  encoder.encode(len, holder.get());
+
+  const char* data = holder.get();
+
+  // when
+  // then
+  EXPECT_THROW(testee.feed(data, remaining), EnvoyException);
+}
+
+// === ARRAY BUFFER ============================================================
+
+TEST(ArrayBuffer, ShouldConsumeCorrectAmountOfData) {
+  const NULLABLE_ARRAY<STRING> value{{"aaa", "bbbbb", "cc", "d", "e", "ffffffff"}};
+  serializeThenDeserializeAndCheckEquality<ArrayBuffer<STRING, StringBuffer>>(value);
+}
+
+TEST(ArrayBuffer, ShouldThrowOnInvalidLength) {
+  // given
+  ArrayBuffer<std::string, StringBuffer> testee;
+
+  uint64_t remaining = 1024;
+  const std::unique_ptr<char[]> holder = std::unique_ptr<char[]>(new char[remaining]);
+
+  INT32 len = -2; // -1 is OK for ARRAY
+  encoder.encode(len, holder.get());
+
+  const char* data = holder.get();
+
+  // when
+  // then
+  EXPECT_THROW(testee.feed(data, remaining), EnvoyException);
+}
+
+// === COMPOSITE BUFFER ========================================================
+
+struct CompositeBufferResult {
+  STRING field1_;
+  NULLABLE_ARRAY<INT32> field2_;
+  INT16 field3_;
+
+  size_t encode(char* dst, Encoder& encoder) const {
+    size_t written{0};
+    written += encoder.encode(field1_, dst + written);
+    written += encoder.encode(field2_, dst + written);
+    written += encoder.encode(field3_, dst + written);
+    return written;
+  }
+};
+
+bool operator==(const CompositeBufferResult& lhs, const CompositeBufferResult& rhs) {
+  return (lhs.field1_ == rhs.field1_) && (lhs.field2_ == rhs.field2_) &&
+         (lhs.field3_ == rhs.field3_);
+}
+
+typedef CompositeBuffer<CompositeBufferResult, StringBuffer, ArrayBuffer<INT32, Int32Buffer>,
+                        Int16Buffer>
+    TestCompositeBuffer;
+
+TEST(CompositeBuffer, ShouldDeserialize) {
+  const CompositeBufferResult expected{"zzzzz", {{10, 20, 30, 40, 50}}, 1234};
+  serializeThenDeserializeAndCheckEquality<TestCompositeBuffer>(expected);
+}
+
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy