Kafkajs: Connection error with consumer (write EPIPE)

[thomasgosse]

Hi !

I'm using KafkaJS for two things:

• 1: Consume messages from a Kafka topic (A) and send these messages to a Google Pubsub topic (A')
• 2: Consume messages from a Pubsub topic (B) and send these messages to a Kafka topic (B')

My code is running in a Google AppEngine always up and running (min_instance: 1).
Step 1 was working fine, my consumer was running for 2 weeks and still handling messages without being disconnected (except during google maintenance when the appengine was rebooted)

I tried to implement step 2 and weird things started to happen:
2020-03-25T10:00:06.879287Z [Connection] Connection error: write EPIPE 2020-03-25T10:00:42.514502Z [Consumer] Crash: KafkaJSNumberOfRetriesExceeded: Not connected 2020-03-25T10:00:42.592285Z [Consumer] Stopped 2020-03-25T10:00:42.592480Z [Consumer] Restarting the consumer in 12158ms 2020-03-25T10:00:54.779075Z [Consumer] Starting 2020-03-25T10:00:58.477868Z [Connection] Connection timeout 2020-03-25T10:00:58.478020Z [BrokerPool] Failed to connect to broker, reconnecting 2020-03-25T10:01:00.189576Z [Runner] Consumer has joined the group

When the app starts, everything is fine for like 1 hour, and then, this error happen, and everything loop crash for days, until I restart the app...

The code I'm using:
app.txt
kafka-handler.txt

Do you have ever experienced this kind of problem?

Do not hesitate if you need any other information !

Thank you

[Nevon]

Just posting the code samples here to make it easier to review:

app.js

const express = require("express");
const cors = require("cors");
const bodyParser = require("body-parser");
const kafkaHandler = require("./kafka-handler");

const app = express();

kafkaHandler.initKafkaConsumer();
kafkaHandler.initPubsubListen();

app.use(bodyParser.json());
app.use(cors());

app.get("/health", (req, res) => {
  return res.json({ message: "Service is up and running" });
});

const PORT = process.env.PORT || 3000;
app.listen(PORT, () => console.log(`App listening on port ${PORT}!`));

kafka-handler.js

const { Kafka } = require("kafkajs");
const { SchemaRegistry, readAVSC } = require("@kafkajs/confluent-schema-registry");
const { PubSub } = require("@google-cloud/pubsub");
const config = require("config");

const { publishToPubsubTopic, publishMetric } = require("./pubsub");

const kafka = new Kafka({
  brokers: [config.get("KAFKA_HOST")],
  clientId: config.get("KAFKA_APPLICATION_ID"),
  ssl: {
    rejectUnauthorized: true
  },
  sasl: {
    mechanism: config.get("KAFKA_SASL_MECHANISM"),
    username: config.get("KAFKA_SASL_USERNAME"),
    password: config.get("KAFKA_SASL_PASSWORD")
  }
});

const registry = new SchemaRegistry({
  host: config.get("SCHEMA_REGISTRY"),
  auth: {
    username: config.get("SCHEMA_REGISTRY_USERNAME"),
    password: config.get("SCHEMA_REGISTRY_PASSWORD")
  }
});

const consumer = kafka.consumer({ groupId: config.get("KAFKA_GROUP_ID") });
const producer = kafka.producer();

/** **/
/** Listen kafka topic messages to send them to a pubsub topic **/

async function handleMessage(topic, partition, message) {
  const decodedMessage = await registry.decode(message.value);
  console.log(`Message received from topic ${topic}, partition ${partition}`);

  const messageToSend = {
    productID: decodedMessage.productID,
    timestamp: decodedMessage.timestamp
  };
  await publishToPubsubTopic(messageToSend);
}

async function initKafkaConsumer() {
  await consumer.connect();
  await consumer.subscribe({ topic: config.get("TOPIC_IN") });

  console.log(`Ready to consume message on topic ${config.get("TOPIC_IN")}`);

  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      let tags = { component: "kafka-consumer", type: "out_success" };
      try {
        await handleMessage(topic, partition, message);
        await publishMetric("count", tags, 1);
      } catch (e) {
        tags.type = "out_error";
        await publishMetric("count", tags, 1);
        console.log("error when handling kfk msg", e);
      }
    }
  });
}

/** **/
/** Listen to pubsub messages to send them to a kafka TOPIC **/

let schemaCache;
async function getSchema() {
  if (!schemaCache) {
    schemaCache = await registry.register(readAVSC("avro/updated.avsc"));
  }
  return schemaCache;
}

async function initPubsubListen() {
  const pubSubClient = new PubSub();
  const subscription = pubSubClient.subscription(config.get("PUBSUB_OUT_KAFKA_SUBSCRIPTION"));
  await producer.connect();

  const messageHandler = async pubSubMessage => {
    let tags = { component: "pubsub2kafka-function", type: "out_success" };
    try {
      let data;
      try {
        data = JSON.parse(Buffer.from(pubSubMessage.data, "base64").toString("utf-8"));
      } catch (parseError) {
        console.log("Error while parsing data", parseError);
        return;
      }

      tags.productID = data.productID;

      const schema = await getSchema();
      const encodedMessage = await registry.encode(schema.id, data);

      await producer.send({
        topic: config.get("TOPIC_OUT"),
        messages: [{ key: data.productID, value: encodedMessage }]
      });

      pubSubMessage.ack();
      await publishMetric("count", tags, 1);
    } catch (e) {
      console.log(e);
      tags.type = "out_error";
      pubSubMessage.nack();
      await publishMetric("count", tags, 1);
    }
  };

  subscription.on("message", messageHandler);
}

module.exports = { initKafkaConsumer, initPubsubListen };