apache / apache/pulsar-adapters

kafka adaptor can not handle non-partitioned topic

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

Description

**Reproduce**
- create non-paritioned topic
- send message to this topic

**error**
![image](https://user-images.githubusercontent.com/9473606/183640375-517bcea0-fa50-48af-8881-9696fab6d558.png)

**probable reason**
PulsarKafkaConsumer -> poll
```
public ConsumerRecords poll(long timeoutMillis) {
try {
QueueItem item = receivedMessages.poll(timeoutMillis, TimeUnit.MILLISECONDS);
if (item == null) {
return (ConsumerRecords) ConsumerRecords.EMPTY;
}

Map>> records = new HashMap<>();

int numberOfRecords = 0;

while (item != null) {
TopicName topicName = TopicName.get(item.consumer.getTopic());
String topic = topicName.getPartitionedTopicName();
int partition = topicName.isPartitioned() ? topicName.getPartitionIndex() : 0;
Message msg = item.message;
MessageId msgId = msg.getMessageId();
if (msgId instanceof TopicMessageIdImpl) {
msgId = ((TopicMessageIdImpl) msgId).getInnerMessageId();
}
long offset = MessageIdUtils.getOffset(msgId);

TopicPartition tp = new TopicPartition(topic, partition);
if (lastReceivedOffset.get(tp) == null && !unpolledPartitions.contains(tp)) {
log.info("When polling offsets, invalid offsets were detected. Resetting topic partition {}", tp);
resetOffsets(tp);
}

// .. other code

// If no interceptor is provided, interceptors list will an empty list, original ConsumerRecords will be return.
return applyConsumerInterceptorsOnConsume(interceptors, new ConsumerRecords<>(records));
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
```
**int partition = topicName.isPartitioned() ? topicName.getPartitionIndex() : 0;**
This code can not discriminate partitioned-topic or non-paritioned-topic.

Contributor guide

No contributing guide indexed for this repository

Research direction

Start in PulsarKafkaConsumer.poll, focusing on the TopicName partition check and the resulting TopicPartition. Reproduce the issue by creating a non-partitioned topic and sending a message, then verify that polling returns a valid ConsumerRecords result for that topic without the reported error.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
distributed-systems
Issue type
Bug
Difficulty
3/5
Estimated time
1-2 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
45/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.