search

nodefluent / kafka-streams

equivalent to kafka-streams :octopus: for nodejs :sparkles::turtle::rocket::sparkles:
https://nodefluent.github.io/kafka-streams/
MIT License
830 stars 111 forks source link

Calling .consume() is not required in streaming mode. #229

Open BufferUnderrunOK opened 2 years ago

BufferUnderrunOK commented 2 years ago

Having the same issue as https://github.com/nodefluent/kafka-streams/issues/91

Error:

[Nest] 56257  - 08/23/2022, 11:34:17 AM   ERROR Error: Calling .consume() is not required in streaming mode.
    at NConsumer.consume (/home/barrett/projects/kstream-poc/node_modules/.pnpm/sinek@9.0.0/node_modules/sinek/lib/librdkafka/NConsumer.js:502:29)
    at /home/barrett/projects/kstream-poc/node_modules/.pnpm/kafka-streams@5.0.0/node_modules/kafka-streams/lib/client/NativeKafkaClient.js:115:42

config:

const kafkaStreamsOptions: KafkaStreamsConfig = {
  kafkaHost: KAFKA_CONFIG.brokers[0],
  clientName: 'tsstreams',
  options: {
    fromOffset: 'earliest',
  },
  noptions: {
    // needs node-rdkafka installed
    'group.id': 'currently-irrelevant',
    'metadata.broker.list': `${KAFKA_CONFIG.brokers}`,
    'security.protocol': 'sasl_plaintext',
    'sasl.mechanisms': KAFKA_CONFIG.sasl.mechanism.toUpperCase(),
    'sasl.username': (KAFKA_CONFIG.sasl as SASLPlainAuthzOptions).username,
    'sasl.password': (KAFKA_CONFIG.sasl as SASLPlainAuthzOptions).password,
    'log.connection.close': false,
  },
};