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 2019/08/01 09:24:23 UTC
[GitHub] [pulsar-client-go] wolfstudy opened a new issue #39: The message
sent by the go client, the Java client's consumer can't receive it normally.
wolfstudy opened a new issue #39: The message sent by the go client, the Java client's consumer can't receive it normally.
URL: https://github.com/apache/pulsar-client-go/issues/39
#### Expected behavior
The message sent by the go client, the Java client's consumer can receive it normally.
#### Actual behavior
The Java client's consumer can't receive it normally.
#### Steps to reproduce
Producer of Go client:
```
func main() {
client, err := pulsar.NewClient(pulsar.ClientOptions{
URL: "pulsar://localhost:6650",
})
if err != nil {
log.Fatal(err)
}
defer client.Close()
producer, err := client.CreateProducer(pulsar.ProducerOptions{
Topic: "my-topic-1",
})
if err != nil {
log.Fatal(err)
}
defer producer.Close()
ctx := context.Background()
for i := 0; i < 10; i++ {
if err := producer.Send(ctx, &pulsar.ProducerMessage{
Payload: []byte(fmt.Sprintf("hello-%d", i)),
}); err != nil {
log.Fatal(err)
}
}
}
```
Java client receive:
```
public static void main(String[] args) throws PulsarClientException, InterruptedException {
final String topic = "persistent://public/default/my-topic-2";
PulsarClient client = PulsarClient.builder()
.serviceUrl("pulsar://localhost:6650")
.build();
Consumer<byte[]> consumer = client.newConsumer()
.topic(topic)
.subscriptionName(UUID.randomUUID().toString())
.subscribe();
for (int i = 0; i < 10; i++) {
Message<byte[]> message = consumer.receive();
System.out.println(new String(message.getValue(), Charset.defaultCharset()));
}
}
```
The broker error information as follows:
```
17:15:56.392 [pulsar-io-37-6] INFO org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.1:56978] Created subscription on topic persistent://public/default/my-topic-2 / a01fed02-2baf-4617-89e7-611dd08633cc
17:15:56.395 [pulsar-io-37-6] WARN org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.1:56978] Got exception IOException : Connection reset by peer
java.io.IOException: Connection reset by peer
at sun.nio.ch.FileDispatcherImpl.read0(Native Method) ~[?:1.8.0_201]
at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:39) ~[?:1.8.0_201]
at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:223) ~[?:1.8.0_201]
at sun.nio.ch.IOUtil.read(IOUtil.java:192) ~[?:1.8.0_201]
at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:380) ~[?:1.8.0_201]
at io.netty.buffer.PooledUnsafeDirectByteBuf.setBytes(PooledUnsafeDirectByteBuf.java:288) ~[io.netty-netty-all-4.1.32.Final.jar:4.1.32.Final]
at io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:1132) ~[io.netty-netty-all-4.1.32.Final.jar:4.1.32.Final]
at io.netty.channel.socket.nio.NioSocketChannel.doReadBytes(NioSocketChannel.java:347) ~[io.netty-netty-all-4.1.32.Final.jar:4.1.32.Final]
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:148) [io.netty-netty-all-4.1.32.Final.jar:4.1.32.Final]
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:656) [io.netty-netty-all-4.1.32.Final.jar:4.1.32.Final]
at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:591) [io.netty-netty-all-4.1.32.Final.jar:4.1.32.Final]
at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:508) [io.netty-netty-all-4.1.32.Final.jar:4.1.32.Final]
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:470) [io.netty-netty-all-4.1.32.Final.jar:4.1.32.Final]
at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:909) [io.netty-netty-all-4.1.32.Final.jar:4.1.32.Final]
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [io.netty-netty-all-4.1.32.Final.jar:4.1.32.Final]
at java.lang.Thread.run(Thread.java:748) [?:1.8.0_201]
17:15:56.396 [pulsar-io-37-6] INFO org.apache.pulsar.broker.service.ServerCnx - Closed connection from /127.0.0.1:56978
17:15:56.396 [pulsar-io-37-6] INFO org.apache.pulsar.broker.service.AbstractDispatcherSingleActiveConsumer - Removing consumer Consumer{subscription=PersistentSubscription{topic=persistent://public/default/my-topic-2, name=a01fed02-2baf-4617-89e7-611dd08633cc}, consumerId=0, consumerName=04824, address=/127.0.0.1:56978}
17:15:56.499 [pulsar-io-37-7] INFO org.apache.pulsar.broker.service.ServerCnx - New connection from /127.0.0.1:56979
17:15:56.499 [pulsar-io-37-7] INFO org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.1:56979] Subscribing on topic persistent://public/default/my-topic-2 / a01fed02-2baf-4617-89e7-611dd08633cc
17:15:56.500 [pulsar-io-37-7] INFO org.apache.bookkeeper.mledger.impl.ManagedCursorImpl - [public/default/persistent/my-topic-2-a01fed02-2baf-4617-89e7-611dd08633cc] Rewind from 7:10 to 7:0
17:15:56.500 [pulsar-io-37-7] INFO org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://public/default/my-topic-2] There are no replicated subscriptions on the topic
17:15:56.500 [pulsar-io-37-7] INFO org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://public/default/my-topic-2][a01fed02-2baf-4617-89e7-611dd08633cc] Created new subscription for 0
```
#### System configuration
**Pulsar version**: x.y
----------------------------------------------------------------
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