l1x · December 11, 2024 13:37
diff --git a/consumer.js b/consumer.js
 const { Kafka, logLevel, Partitioners, CompressionTypes, CompressionCodecs } =
  require("kafkajs");
 const ZstdCodec = require("@kafkajs/zstd");

 const compressionParams = { level: 1 };
 const decompressionParams = {};
 CompressionCodecs[CompressionTypes.ZSTD] = ZstdCodec(
  compressionParams,
  decompressionParams,
 );

 const kafka = new Kafka({
  clientId: "my-consumer-client",
  brokers: [
    "b-2.kafkadevcluster.qmy2pc.c3.kafka.eu-west-1.amazonaws.com:9092",
    "b-1.kafkadevcluster.qmy2pc.c3.kafka.eu-west-1.amazonaws.com:9092",
  ],
  connectionTimeout: 10000,
  requestTimeout: 15000,
  retry: {
    retries: 10,
    factor: 2,
    maxRetryTime: 60000,
    initialRetryTime: 500,
  },
  logLevel: logLevel.INFO,
  ack: "all",
 });

 const consumer = kafka.consumer({ groupId: "my-consumer-group" });

 const run = async () => {
  // Connect to Kafka broker
  await consumer.connect();
  console.log("Connected to Kafka");

  // Subscribe to a Kafka topic
  await consumer.subscribe({ topic: "test-topic", fromBeginning: true }); // Change topic name

  // Start consuming messages
  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      console.log(`Received message: ${message.value.toString()}`);
      console.log(`From topic: ${topic}, Partition: ${partition}`);
      // Add your custom message processing logic here
    },
  });
 };

 // Start the consumer
 run().catch(console.error);
diff --git a/cppzst.md b/cppzst.md
diff --git a/mise.toml b/mise.toml
 [tasks.fmt]
 description = "FMT"
 run = "deno fmt"

 [tools]
 bun = "latest"
 deno = "latest"
 node = "lts"
diff --git a/package.json b/package.json
 {
  "dependencies": {
    "@kafkajs/zstd": "^0.1.1",
    "kafkajs": "^2.2.4",
    "node-gyp": "^11.0.0"
  }
 }
diff --git a/producer.js b/producer.js
 const { Kafka, logLevel, Partitioners } = require("kafkajs");

 const kafka = new Kafka({
  clientId: "my-consumer-client",
  brokers: [
    "b-2.kafkadevcluster.qmy2pc.c3.kafka.eu-west-1.amazonaws.com:9092",
    "b-1.kafkadevcluster.qmy2pc.c3.kafka.eu-west-1.amazonaws.com:9092",
  ],
  connectionTimeout: 10000,
  requestTimeout: 15000,
  retry: {
    retries: 10,
    factor: 2,
    maxRetryTime: 60000,
    initialRetryTime: 500,
  },
  logLevel: logLevel.INFO,
  ack: "all",
 });

 const admin = kafka.admin();

 await admin.connect();

 const topics = await admin.listTopics();

 if (topics.indexOf("test-topic") == -1) {
  const createTopic = await admin.createTopics({
    topics: [
      {
        topic: "test-topic",
        replicationFactor: 2,
        numPartitions: 2,
      },
    ],
  });
  createTopic
    ? console.log("Topic CREATED!")
    : console.log("TOPIC CREATION ERROR!");
 }

 const producer = kafka.producer({
  createPartitioner: Partitioners.DefaultPartitioner,
 });

 const produceMessage = async () => {
  try {
    await producer.connect();
    console.log("Producer connected.");
    const promises = [...Array(100).keys()].map(async (element) => {
      const result = await producer.send({
        topic: "test-topic", // The Kafka topic
        messages: [
          { value: `Hello, Kafka! This is event: ${element}` },
        ],
      });
      console.log("Message sent:", result);
    });
    await Promise.all(promises);
  } catch (error) {
    console.error("Error sending message:", error);
  } finally {
    console.log("Finally start");
    await producer.disconnect();
    console.log("Finally end");
  }
 };

 await produceMessage();
 console.log("The end");

 process.exit(0);
	const { Kafka, logLevel, Partitioners, CompressionTypes, CompressionCodecs } =
	require("kafkajs");
	const ZstdCodec = require("@kafkajs/zstd");

	const compressionParams = { level: 1 };
	const decompressionParams = {};
	CompressionCodecs[CompressionTypes.ZSTD] = ZstdCodec(
	compressionParams,
	decompressionParams,
	);

	const kafka = new Kafka({
	clientId: "my-consumer-client",
	brokers: [
	"b-2.kafkadevcluster.qmy2pc.c3.kafka.eu-west-1.amazonaws.com:9092",
	"b-1.kafkadevcluster.qmy2pc.c3.kafka.eu-west-1.amazonaws.com:9092",
	],
	connectionTimeout: 10000,
	requestTimeout: 15000,
	retry: {
	retries: 10,
	factor: 2,
	maxRetryTime: 60000,
	initialRetryTime: 500,
	},
	logLevel: logLevel.INFO,
	ack: "all",
	});

	const consumer = kafka.consumer({ groupId: "my-consumer-group" });

	const run = async () => {
	// Connect to Kafka broker
	await consumer.connect();
	console.log("Connected to Kafka");

	// Subscribe to a Kafka topic
	await consumer.subscribe({ topic: "test-topic", fromBeginning: true }); // Change topic name

	// Start consuming messages
	await consumer.run({
	eachMessage: async ({ topic, partition, message }) => {
	console.log(`Received message: ${message.value.toString()}`);
	console.log(`From topic: ${topic}, Partition: ${partition}`);
	// Add your custom message processing logic here
	},
	});
	};

	// Start the consumer
	run().catch(console.error);
	[tasks.fmt]
	description = "FMT"
	run = "deno fmt"

	[tools]
	bun = "latest"
	deno = "latest"
	node = "lts"
	{
	"dependencies": {
	"@kafkajs/zstd": "^0.1.1",
	"kafkajs": "^2.2.4",
	"node-gyp": "^11.0.0"
	}
	}
	const { Kafka, logLevel, Partitioners } = require("kafkajs");

	const kafka = new Kafka({
	clientId: "my-consumer-client",
	brokers: [
	"b-2.kafkadevcluster.qmy2pc.c3.kafka.eu-west-1.amazonaws.com:9092",
	"b-1.kafkadevcluster.qmy2pc.c3.kafka.eu-west-1.amazonaws.com:9092",
	],
	connectionTimeout: 10000,
	requestTimeout: 15000,
	retry: {
	retries: 10,
	factor: 2,
	maxRetryTime: 60000,
	initialRetryTime: 500,
	},
	logLevel: logLevel.INFO,
	ack: "all",
	});

	const admin = kafka.admin();

	await admin.connect();

	const topics = await admin.listTopics();

	if (topics.indexOf("test-topic") == -1) {
	const createTopic = await admin.createTopics({
	topics: [
	{
	topic: "test-topic",
	replicationFactor: 2,
	numPartitions: 2,
	},
	],
	});
	createTopic
	? console.log("Topic CREATED!")
	: console.log("TOPIC CREATION ERROR!");
	}

	const producer = kafka.producer({
	createPartitioner: Partitioners.DefaultPartitioner,
	});

	const produceMessage = async () => {
	try {
	await producer.connect();
	console.log("Producer connected.");
	const promises = [...Array(100).keys()].map(async (element) => {
	const result = await producer.send({
	topic: "test-topic", // The Kafka topic
	messages: [
	{ value: `Hello, Kafka! This is event: ${element}` },
	],
	});
	console.log("Message sent:", result);
	});
	await Promise.all(promises);
	} catch (error) {
	console.error("Error sending message:", error);
	} finally {
	console.log("Finally start");
	await producer.disconnect();
	console.log("Finally end");
	}
	};

	await produceMessage();
	console.log("The end");

	process.exit(0);