apache / apache/pulsar-adapters

Bug in doCommitOffsets of PulsarKafkaConsumer [Ver: 2.11.0]

Open
#48 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
Java
Stars
24
Forks
30
PR merge metrics
No merged PRs in 30d

Description

**Describe the bug**

The 2.11.0 version of pulsar-client-kafka-compat has bug that failes commitSync always. This seems to have been introduced in this PR - https://github.com/apache/pulsar-adapters/pull/37.

```java
if (consumer instanceof MultiTopicsConsumerImpl) {
msgId = new TopicMessageIdImpl(topicPartition.topic(), tp.topic(), msgId);
}
```

Now in MultiTopicsConsumerImpl map of the consumer with key as a topic for non-partitioned topic and topicPartitionName for the partitioned topic.

TopicMessageIdImpl takes the partition name as the first argument, but here topic name is passed. In the case of the partitioned topic, the value should have been suffixed with -partition-. But what is passed is just the topic name for which there is no consumer created in MultiTopicConsumerImpl resulting

```java
Caused by: org.apache.pulsar.client.api.PulsarClientException$NotConnectedException: Not connected to broker
at org.apache.pulsar.client.impl.MultiTopicsConsumerImpl.doAcknowledge(MultiTopicsConsumerImpl.java:503)
at org.apache.pulsar.client.impl.ConsumerBase.doAcknowledgeWithTxn(ConsumerBase.java:650)
at org.apache.pulsar.client.impl.ConsumerBase.acknowledgeCumulativeAsync(ConsumerBase.java:616)
at org.apache.pulsar.client.impl.ConsumerBase.acknowledgeCumulativeAsync(ConsumerBase.java:601)
at org.apache.kafka.clients.consumer.PulsarKafkaConsumer.lambda$doCommitOffsets$9(PulsarKafkaConsumer.java:526)
at java.util.Collections$SingletonMap.forEach(Collections.java:4912)
at org.apache.kafka.clients.consumer.PulsarKafkaConsumer.doCommitOffsets(PulsarKafkaConsumer.java:517)
```

Contributor guide

No contributing guide indexed for this repository

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.