Kafkajs: Include snappy compression (or working example)

[pimpelsang]

Hi,

Tried to do it myself according to example from readme.md, but hitting "Invalid input" error.

So it would be usefull to have either built in support (auto-detecting snappy module presence if don't want to include c++ dependency) or at least working example in /examples folder.

My code:

//snappy-compression.js
var snappy = require('snappy');

const { promisify } = require('util');
const snappyCompress = promisify(snappy.compress)
const snappyDecompress = promisify(snappy.uncompress)

module.exports = {
  async compress(encoder) {
    return await snappyCompress(encoder.buffer)
  },

  async decompress(buffer) {
      return await snappyDecompress(buffer)
  }
}

//index.js
const { Kafka, logLevel, CompressionTypes, CompressionCodecs } = require('kafkajs')

CompressionCodecs[CompressionTypes.Snappy] = () => require('./snappy-compression');

// Create the client with the broker list
const kafka = new Kafka({...

I see data arriving on INFO level logs, snappy decompress run but unable to decompress the buffer.

{"level":"DEBUG","timestamp":"2018-08-24T19:39:21.578Z","logger":"kafkajs","message":"[Connection] Response Fetch(key: 1, version: 3)","broker":"IP.IP.IP.IP:9092","clientId":"kafkajs-app","error":"Invalid input","correlationId":3,"payload":{"type":"Buffer","data":[0,0,0,0,0,0,0,1,0,19,100,101,98,101,122,105,117,109,45,104,101,97,114...

Using kafka 1.1, producing data with java client

[tulios]

Hi @pimpelsang, I naively assumed that the library would be enough to deal with the snappy decompression, but I learned today about something called frame support, which is used by Kafka. Here is a working example of the snappy codec, I'll add it to the examples folder. Maybe create a kafkajs-snappy package.

https://gist.github.com/tulios/0a6010e51c42560a9ec47c002eb2481a

[pimpelsang]

Confirmed your code works. Will you create the package? I can create it as well, but feel like taking credit for your work :)

[tulios]

I moved my gist to an NPM package
https://www.npmjs.com/package/kafkajs-snappy

I'll update the readme later