const { KafkaJSNotImplemented } = require('../../../errors')

const Types = {
  None: 0,
  GZIP: 1,
  Snappy: 2,
  LZ4: 3,
}

const Codecs = {
  [Types.GZIP]: () => require('./gzip'),
  [Types.Snappy]: () => {
    throw new KafkaJSNotImplemented('Snappy compression not implemented')
  },
  [Types.LZ4]: () => {
    throw new KafkaJSNotImplemented('LZ4 compression not implemented')
  },
}

const lookupCodec = type => (Codecs[type] ? Codecs[type]() : null)
const lookupCodecByAttributes = attributes => {
  const codec = Codecs[attributes & 0x3]
  return codec ? codec() : null
}

module.exports = {
  Types,
  Codecs,
  lookupCodec,
  lookupCodecByAttributes,
}

@@ -55,16 +55,85 @@
 rors')%0A%0A
+const MESSAGE_CODEC_MASK = 0x3%0Aconst RECORD_BATCH_CODEC_MASK = 0x07%0A%0A
 const Ty
@@ -604,19 +604,188 @@
 butes & 
-0x3
+MESSAGE_CODEC_MASK%5D%0A  return codec ? codec() : null%0A%7D%0Aconst lookupCodecByRecordBatchAttributes = attributes =%3E %7B%0A  const codec = Codecs%5Battributes & RECORD_BATCH_CODEC_MASK
 %5D%0A  retu
@@ -866,16 +866,16 @@
 pCodec,%0A
-
   lookup
@@ -893,10 +893,48 @@
 ibutes,%0A
+  lookupCodecByRecordBatchAttributes,%0A
 %7D%0A

Add support for Record Batch compression codec flag