text: 'use strict';
  
  var Protocol = require('./index');
  var globals  = require('./globals');
  
  // https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol
  
  /////////////////////////////
  // OFFSET COMMIT/FETCH API //
  /////////////////////////////
  
  // group coordinator request and response are defined in group_membership.js
  
  // v0
  Protocol.define('OffsetCommitRequestV0PartitionItem', {
    write: function (data) { // { partition, offset, metadata }
      this
        .Int32BE(data.partition)
        .Int64BE(data.offset)
        .string(data.metadata);
    },
  });
  
  Protocol.define('OffsetCommitRequestV0TopicItem', {
    write: function (data) { // { topicName, partitions }
      this
        .string(data.topicName)
        .array(data.partitions, this.OffsetCommitRequestV0PartitionItem);
    },
  });
  
  Protocol.define('OffsetCommitRequestV0', {
    write: function (data) { // { groupId, topics }
      this
        .RequestHeader({
          apiKey: globals.API_KEYS.OffsetCommitRequest,
          apiVersion: 0,
          correlationId: data.correlationId,
          clientId: data.clientId,
        })
        .string(data.groupId)
        .array(data.topics, this.OffsetCommitRequestV0TopicItem);
    },
  });
  
  // v1
  /* istanbul ignore next */
  Protocol.define('OffsetCommitRequestV1PartitionItem', {
    write: function (data) { // { partition, offset, timestamp, metadata }
      this
        .Int32BE(data.partition)
        .Int64BE(data.offset)
        .Int64BE(data.timestamp)
        .string(data.metadata);
    },
  });
  
  /* istanbul ignore next */
  Protocol.define('OffsetCommitRequestV1TopicItem', {
    write: function (data) { // { topicName, partitions }
      this
        .string(data.topicName)
        .array(data.partitions, this.OffsetCommitRequestV1PartitionItem);
    },
  });
  
  /* istanbul ignore next */
  Protocol.define('OffsetCommitRequestV1', {
    write: function (data) { // { groupId, generationId, memberId,  topics }
      this
        .RequestHeader({
          apiKey: globals.API_KEYS.OffsetCommitRequest,
          apiVersion: 1,
          correlationId: data.correlationId,
          clientId: data.clientId,
        })
        .string(data.groupId)
        .string(data.generationId)
        .string(data.memberId)
        .array(data.topics, this.OffsetCommitRequestV1TopicItem);
    },
  });
  
  // v2
  Protocol.define('OffsetCommitRequestV2', {
    write: function (data) { // { groupId, generationId, memberId, retentionTime, topics }
      this
        .RequestHeader({
          apiKey: globals.API_KEYS.OffsetCommitRequest,
          apiVersion: 2,
          correlationId: data.correlationId,
          clientId: data.clientId,
        })
        .string(data.groupId)
        .Int32BE(data.generationId)
        .string(data.memberId)
        .Int64BE(data.retentionTime)
        .array(data.topics, this.OffsetCommitRequestV0TopicItem); // same as in v0
    },
  });
  
  
  Protocol.define('OffsetCommitResponseTopicItem', {
    read: function () {
      this
        .string('topicName')
        .array('partitions', this.OffsetCommitResponsePartitionItem);
    },
  });
  
  Protocol.define('OffsetCommitResponsePartitionItem', {
    read: function () {
      this
        .Int32BE('partition')
        .ErrorCode('error');
    },
  });
  
  // v0, v1 and v2
  Protocol.define('OffsetCommitResponse', {
    read: function () {
      this
        .Int32BE('correlationId')
        .array('topics', this.OffsetCommitResponseTopicItem);
    },
  });
  
  
  Protocol.define('OffsetFetchRequestTopicItem', {
    write: function (data) { // { topicName, partitions }
      this
        .string(data.topicName)
        .array(data.partitions, this.Int32BE);
    },
  });
  
  Protocol.define('OffsetFetchRequest', {
    write: function (data) { // { apiVersion, groupId, topics }
      this
        .RequestHeader({
          apiKey: globals.API_KEYS.OffsetFetchRequest,
          apiVersion: data.apiVersion || 0,
          correlationId: data.correlationId,
          clientId: data.clientId,
        })
        .string(data.groupId)
        .array(data.topics, this.OffsetFetchRequestTopicItem);
    },
  });
  
  Protocol.define('OffsetFetchResponseTopicItem', {
    read: function () {
      this
        .string('topicName')
        .array('partitions', this.OffsetFetchResponsePartitionItem);
    },
  });
  
  Protocol.define('OffsetFetchResponsePartitionItem', {
    read: function () {
      this
        .Int32BE('partition')
        .KafkaOffset('offset')
        .string('metadata')
        .ErrorCode('error');
    },
  });
  
  Protocol.define('OffsetFetchResponse', {
    read: function () {
      this
        .Int32BE('correlationId')
        .array('topics', this.OffsetFetchResponseTopicItem);
    },
  });
  