You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2020/03/15 13:43:32 UTC
[GitHub] [pulsar] hailin0 opened a new issue #6540: java-client:2.4.1
consumer message missing
hailin0 opened a new issue #6540: java-client:2.4.1 consumer message missing
URL: https://github.com/apache/pulsar/issues/6540
**Describe the bug**
java-client:2.4.1 consumer message missing when SubscriptionType "Key_Shared"
**To Reproduce**
Steps to reproduce the behavior:
a. pulsar server: 2.4.1
b. pulsar client: java client 2.4.1
c. consumer use SubscriptionType "Key_Shared"
1. Create partitioned topic
./pulsar-admin topics create-partitioned-topic persistent://public/default/test --partitions 3
2. Producer message
```java
PulsarClient client = PulsarClient.builder()
.serviceUrl("pulsar://localhost:6650")
.build();
Producer<byte[]> producer = client.newProducer()
.topic("persistent://public/default/test")
.batcherBuilder(BatcherBuilder.KEY_BASED)
.enableBatching(true)
.create();
for (int i=1;i<=18;i++) {
producer.newMessage()
.key(String.format("key-%s", i))
.value("value-1".getBytes())
.send();
producer.newMessage()
.key(String.format("key-%s", i))
.value("value-2".getBytes())
.send();
producer.newMessage()
.key(String.format("key-%s", i))
.value("value-3".getBytes())
.send();
}
```
2. Consumer message
```java
ExecutorService executorService = Executors.newFixedThreadPool(9);
for (int i=0;i<9;i++) {
executorService.execute(new Runnable() {
@Override
public void run() {
try {
PulsarClient client = PulsarClient.builder()
.serviceUrl("pulsar://localhost:6650")
.build();
Consumer consumer = client.newConsumer()
.topic("persistent://public/default/test")
.subscriptionName("my-subscription")
.subscriptionType(SubscriptionType.Key_Shared)
.subscribe();
while (true) {
Message msg = consumer.receive(1, TimeUnit.SECONDS);
if (msg == null) {
continue;
}
try {
System.out.println(String.format("%s : %s : %s",
Thread.currentThread().getName(),
consumer.getConsumerName(),
msg.getKey()));
consumer.acknowledge(msg);
} catch (Exception e) {
consumer.negativeAcknowledge(msg);
}
if ("key-10".equals(msg.getKey())) {
System.out.println(String.format("%s : %s : %s",
Thread.currentThread().getName(),
consumer.getConsumerName(),
" -关闭- "));
consumer.close();
client.close();
break;
}
}
} catch (Throwable e) {
e.printStackTrace();
}
}
});
}
```
4. See error
I see message missing of parttion key "key-10" .
If you haven't see, please try a few more times. It happens occasionally
```log
pool-1-thread-8 : 37ab0 : key-1
pool-1-thread-8 : 37ab0 : key-1
pool-1-thread-8 : 37ab0 : key-1
pool-1-thread-8 : 37ab0 : key-2
pool-1-thread-8 : 37ab0 : key-2
pool-1-thread-8 : 37ab0 : key-2
pool-1-thread-1 : de91e : key-3
pool-1-thread-1 : de91e : key-3
pool-1-thread-1 : de91e : key-3
pool-1-thread-7 : 32150 : key-4
pool-1-thread-7 : 32150 : key-4
pool-1-thread-7 : 32150 : key-4
pool-1-thread-3 : 67979 : key-5
pool-1-thread-3 : 67979 : key-5
pool-1-thread-3 : 67979 : key-5
pool-1-thread-1 : de91e : key-6
pool-1-thread-1 : de91e : key-6
pool-1-thread-1 : de91e : key-6
pool-1-thread-2 : 58d49 : key-7
pool-1-thread-2 : 58d49 : key-7
pool-1-thread-2 : 58d49 : key-7
pool-1-thread-7 : 32150 : key-8
pool-1-thread-7 : 32150 : key-8
pool-1-thread-7 : 32150 : key-8
pool-1-thread-1 : de91e : key-9
pool-1-thread-1 : de91e : key-9
pool-1-thread-1 : de91e : key-9
pool-1-thread-7 : 32150 : key-10
pool-1-thread-7 : 32150 : -关闭-
pool-1-thread-2 : 58d49 : key-10
pool-1-thread-2 : 58d49 : -关闭-
pool-1-thread-3 : 67979 : key-11
pool-1-thread-3 : 67979 : key-11
pool-1-thread-3 : 67979 : key-11
pool-1-thread-8 : 37ab0 : key-12
pool-1-thread-8 : 37ab0 : key-12
pool-1-thread-8 : 37ab0 : key-12
pool-1-thread-3 : 67979 : key-13
pool-1-thread-3 : 67979 : key-13
pool-1-thread-3 : 67979 : key-13
pool-1-thread-5 : acffc : key-14
pool-1-thread-5 : acffc : key-14
pool-1-thread-5 : acffc : key-14
pool-1-thread-1 : de91e : key-15
pool-1-thread-1 : de91e : key-15
pool-1-thread-1 : de91e : key-15
pool-1-thread-9 : b2bbf : key-16
pool-1-thread-9 : b2bbf : key-16
pool-1-thread-9 : b2bbf : key-16
pool-1-thread-6 : 4bfe1 : key-17
pool-1-thread-6 : 4bfe1 : key-17
pool-1-thread-6 : 4bfe1 : key-17
pool-1-thread-5 : acffc : key-18
pool-1-thread-5 : acffc : key-18
pool-1-thread-5 : acffc : key-18
```
**Expected behavior**
A clear and concise description of what you expected to happen.
**Screenshots**
If applicable, add screenshots to help explain your problem.
**Desktop (please complete the following information):**
- server：ubuntu
- client：macos
**Additional context**
Add any other context about the problem here.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [pulsar] codelipenghui commented on issue #6540: java-client:2.4.1
consumer message missing
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on issue #6540: java-client:2.4.1 consumer message missing
URL: https://github.com/apache/pulsar/issues/6540#issuecomment-599448957
Looks related to #5894 if it happens occasionally.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services