[GitHub] [pulsar] sijie commented on a change in pull request #4911: expose getLastMessageId method in ConsumerImpl
sijie commented on a change in pull request #4911: expose getLastMessageId method in ConsumerImpl URL: https://github.com/apache/pulsar/pull/4911#discussion_r313229980 ## File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiMessageIdImpl.java ## @@ -0,0 +1,111 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.client.impl; + +import java.util.Map; +import java.util.Map.Entry; +import java.util.Objects; +import lombok.Getter; +import org.apache.pulsar.client.api.MessageId; +import sun.reflect.generics.reflectiveObjects.NotImplementedException; + +/** + * A MessageId implementation that contains a map of . + * This is useful when MessageId is need for partition/multi-topics/pattern consumer. + * e.g. seek(), ackCumulative(), getLastMessageId(). + */ +public class MultiMessageIdImpl implements MessageId { +@Getter +private Map map; + +MultiMessageIdImpl(Map map) { +this.map = map; +} + +// TODO: Add support for Serialization and Deserialization +// https://github.com/apache/pulsar/issues/4940 +@Override +public byte[] toByteArray() { +throw new NotImplementedException(); +} + +@Override +public int hashCode() { +return Objects.hash(map); +} + +// If all messageId in map are same size, and all bigger/smaller than the other, return valid value. +@Override +public int compareTo(MessageId o) { +if (!(o instanceof MultiMessageIdImpl)) { +throw new IllegalArgumentException( +"expected MultiMessageIdImpl object. Got instance of " + o.getClass().getName()); +} + +MultiMessageIdImpl other = (MultiMessageIdImpl) o; +Map otherMap = other.getMap(); + +if ((map == null || map.isEmpty()) && (otherMap == null || otherMap.isEmpty())) { +return 0; +} + +if (otherMap == null || map == null || otherMap.size() != map.size()) { +throw new IllegalArgumentException("Current size and other size not equals"); +} + +int result = 0; +for (Entry entry : map.entrySet()) { +MessageId otherMessage = otherMap.get(entry.getKey()); +if (otherMessage == null) { +throw new IllegalArgumentException( +"Other MessageId not have topic " + entry.getKey()); +} + +int currentResult = entry.getValue().compareTo(otherMessage); +if (result == 0) { +result = currentResult; Review comment: I think the logic here is problematic. This resets the result back to latest result. a MultipleMessageId is larger than the other one is when all the partitions' messageId is larger or equals to the messageIds, no? 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: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [pulsar] jiazhai commented on a change in pull request #4911: expose getLastMessageId method in ConsumerImpl
jiazhai commented on a change in pull request #4911: expose getLastMessageId method in ConsumerImpl URL: https://github.com/apache/pulsar/pull/4911#discussion_r313306254 ## File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiMessageIdImpl.java ## @@ -0,0 +1,111 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.client.impl; + +import java.util.Map; +import java.util.Map.Entry; +import java.util.Objects; +import lombok.Getter; +import org.apache.pulsar.client.api.MessageId; +import sun.reflect.generics.reflectiveObjects.NotImplementedException; + +/** + * A MessageId implementation that contains a map of . + * This is useful when MessageId is need for partition/multi-topics/pattern consumer. + * e.g. seek(), ackCumulative(), getLastMessageId(). + */ +public class MultiMessageIdImpl implements MessageId { +@Getter +private Map map; + +MultiMessageIdImpl(Map map) { +this.map = map; +} + +// TODO: Add support for Serialization and Deserialization +// https://github.com/apache/pulsar/issues/4940 +@Override +public byte[] toByteArray() { +throw new NotImplementedException(); +} + +@Override +public int hashCode() { +return Objects.hash(map); +} + +// If all messageId in map are same size, and all bigger/smaller than the other, return valid value. +@Override +public int compareTo(MessageId o) { +if (!(o instanceof MultiMessageIdImpl)) { +throw new IllegalArgumentException( +"expected MultiMessageIdImpl object. Got instance of " + o.getClass().getName()); +} + +MultiMessageIdImpl other = (MultiMessageIdImpl) o; +Map otherMap = other.getMap(); + +if ((map == null || map.isEmpty()) && (otherMap == null || otherMap.isEmpty())) { +return 0; +} + +if (otherMap == null || map == null || otherMap.size() != map.size()) { +throw new IllegalArgumentException("Current size and other size not equals"); +} + +int result = 0; +for (Entry entry : map.entrySet()) { +MessageId otherMessage = otherMap.get(entry.getKey()); +if (otherMessage == null) { +throw new IllegalArgumentException( +"Other MessageId not have topic " + entry.getKey()); +} + +int currentResult = entry.getValue().compareTo(otherMessage); +if (result == 0) { +result = currentResult; Review comment: 1. result == 0 && currentResult ==0, then this reset has no effect. 2. result ==0 && currentResult !=0, then we keep the non-zero result, this is the result that may be finally returned. 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: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [pulsar] kimcs commented on issue #4912: Java client Reader readNext and hasMessageAvailable not working when attempting to find last message in topic
kimcs commented on issue #4912: Java client Reader readNext and hasMessageAvailable not working when attempting to find last message in topic URL: https://github.com/apache/pulsar/issues/4912#issuecomment-520805615 Currently I'm using the following workaround to be able to read the last message of a topic: Use the pulsar-admin-client to get the last-message-id of a topic (-1 means empty topic), then use pulsar-client Consumer.seek to that id and then the next receive call will get the last message of the topic. This is not an ideal solution as it also relies on the pulsar-admin client. 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: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [pulsar-client-go] jiazhai commented on a change in pull request #43: Support partition consumer receive async and fix batch logic
jiazhai commented on a change in pull request #43: Support partition consumer receive async and fix batch logic URL: https://github.com/apache/pulsar-client-go/pull/43#discussion_r313375268 ## File path: pulsar/impl_partition_consumer.go ## @@ -274,30 +320,45 @@ func (pc *partitionConsumer) ReceiveAsync(ctx context.Context, msgs chan<- Consu case tmpMsg, ok := <-pc.subQueue: if ok { msgs <- tmpMsg - id := &pb.MessageIdData{} - err := proto.Unmarshal(tmpMsg.ID().Serialize(), id) + + err := pc.messageProcessed(tmpMsg.ID(), receivedSinceFlow) if err != nil { - pc.log.WithError(err).Errorf("unserialize message id error:%s", err.Error()) return err } - if pc.unAckTracker != nil { - pc.unAckTracker.Add(id) - } - receivedSinceFlow++ - if receivedSinceFlow >= highwater { - if err := pc.internalFlow(receivedSinceFlow); err != nil { - pc.log.Errorf("Send Flow cmd error:%s", err.Error()) - return err - } - receivedSinceFlow = 0 - } continue } + break case <-ctx.Done(): return ctx.Err() } } +} + +func (pc *partitionConsumer) ReceiveAsyncWithCallback(ctx context.Context, callback func(msg Message, err error)) { + highwater := uint32(math.Max(float64(cap(pc.options.MessageChannel)/2), 1)) + + // request half the buffer's capacity + if err := pc.internalFlow(highwater); err != nil { Review comment: try to do it in background 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: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [pulsar-client-go] jiazhai commented on a change in pull request #43: Support partition consumer receive async and fix batch logic
jiazhai commented on a change in pull request #43: Support partition consumer receive async and fix batch logic URL: https://github.com/apache/pulsar-client-go/pull/43#discussion_r313375713 ## File path: pulsar/impl_partition_consumer.go ## @@ -238,32 +252,64 @@ func (pc *partitionConsumer) internalUnsubscribe(unsub *handleUnsubscribe) { unsub.waitGroup.Done() } -func (pc *partitionConsumer) Receive(ctx context.Context) (Message, error) { - select { - case <-ctx.Done(): - return nil, ctx.Err() - case cm, ok := <-pc.subQueue: - if ok { - id := &pb.MessageIdData{} - err := proto.Unmarshal(cm.ID().Serialize(), id) - if err != nil { - pc.log.WithError(err).Errorf("unserialize message id error:%s", err.Error()) - return nil, err - } - if pc.unAckTracker != nil { - pc.unAckTracker.Add(id) - } - return cm.Message, nil +func (pc *partitionConsumer) trackMessage(msgID MessageID) error { + id := &pb.MessageIdData{} + err := proto.Unmarshal(msgID.Serialize(), id) + if err != nil { + pc.log.WithError(err).Errorf("unserialize message id error:%s", err.Error()) + return err + } + if pc.unAckTracker != nil { + pc.unAckTracker.Add(id) + } + return nil +} + +func (pc *partitionConsumer) increaseAvailablePermits(receivedSinceFlow uint32) error { + highwater := uint32(math.Max(float64(cap(pc.options.MessageChannel)/2), 1)) + if receivedSinceFlow >= highwater { + if err := pc.internalFlow(receivedSinceFlow); err != nil { + pc.log.Errorf("Send Flow cmd error:%s", err.Error()) + return err } - return nil, newError(ResultConnectError, "receive queue closed") + receivedSinceFlow = 0 + } + return nil +} + +func (pc *partitionConsumer) messageProcessed(msgID MessageID, receivedSinceFlow uint32) error { + err := pc.trackMessage(msgID) + if err != nil { + return err + } + receivedSinceFlow++ + + err = pc.increaseAvailablePermits(receivedSinceFlow) + if err != nil { + return err } + + return nil +} + +func (pc *partitionConsumer) Receive(ctx context.Context) (message Message, err error) { + wg := &sync.WaitGroup{} + wg.Add(1) + pc.ReceiveAsyncWithCallback(ctx, func(msg Message, e error) { + message = msg + err = e + wg.Done() + }) + wg.Wait() + + return message, err } func (pc *partitionConsumer) ReceiveAsync(ctx context.Context, msgs chan<- ConsumerMessage) error { - highwater := uint32(math.Max(float64(cap(pc.options.MessageChannel)/2), 1)) + highWater := uint32(math.Max(float64(cap(pc.options.MessageChannel)/2), 1)) // request half the buffer's capacity - if err := pc.internalFlow(highwater); err != nil { + if err := pc.internalFlow(highWater); err != nil { Review comment: also here, in background 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: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [pulsar] tuteng commented on issue #4918: [dashboard] fix peek parse message error
tuteng commented on issue #4918: [dashboard] fix peek parse message error URL: https://github.com/apache/pulsar/pull/4918#issuecomment-520874584 lgtm 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: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [pulsar] codelipenghui commented on issue #4621: [PIP-38] Support batch receive in java client.
codelipenghui commented on issue #4621: [PIP-38] Support batch receive in java client. URL: https://github.com/apache/pulsar/pull/4621#issuecomment-520881629 run java8 tests 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: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [pulsar] mingfang commented on issue #3121: Several issues with the kubernetes generic example
mingfang commented on issue #3121: Several issues with the kubernetes generic example URL: https://github.com/apache/pulsar/issues/3121#issuecomment-520952751 I'm still seeing this problem with 2.4.0. I'm deploying bookie as stateful set and just using in container storage for testing. The error seems to be related to zookeeper info and not related to storage. 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: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [pulsar] marcusalmeida opened a new pull request #4946: [pulsar-functions][python-examples] Fixing python function example custom_object_function.py
marcusalmeida opened a new pull request #4946: [pulsar-functions][python-examples] Fixing python function example custom_object_function.py URL: https://github.com/apache/pulsar/pull/4946 ### Motivation Fixing Python examples for pulsar-functions custom_object_function.py ### Modifications Adding conversion of "str" to "bytes" using the methods "encode" and "decode" from string. To perform the tests run: ```sh $ bin/pulsar-admin functions create \ --py $PULSAR_HOME/examples/python-examples/custom_object_function.py \ --tenant public \ --namespace default \ --name custom_object \ --inputs input-topic \ --custom-serde-inputs '{"input-topic": "custom_object_function.CustomSerDe"}' \ --output output-topic \ --output-serde-classname custom_object_function.CustomSerDe \ --classname custom_object_function.CustomObjectFunction ``` And trigger the function using the CLI "functions trigger" : ```sh $ pulsar-admin functions trigger \ --name custom \ --tenant public \ --namespace default \ --trigger-value "1,2" ``` And should return: 12, 26 ### Does this pull request potentially affect one of the following parts: *If `yes` was chosen, please highlight the changes* - Dependencies (does it add or upgrade a dependency): no - The public API: no - The schema: no - The default values of configurations: no - The wire protocol: no - The rest endpoints: no - The admin cli options: no - Anything that affects deployment: no ### Documentation - Does this pull request introduce a new feature? no 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: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [pulsar] massakam commented on issue #4945: [pulsar-websocket] Do not log stack trace with warn log in websocket proxy
massakam commented on issue #4945: [pulsar-websocket] Do not log stack trace with warn log in websocket proxy URL: https://github.com/apache/pulsar/pull/4945#issuecomment-521066796 rerun cpp tests 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: us...@infra.apache.org With regards, Apache Git Services
[pulsar] branch master updated: [dashboard] fix peek parse message error (#4918)
This is an automated email from the ASF dual-hosted git repository. sijie pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/pulsar.git The following commit(s) were added to refs/heads/master by this push: new a8b57c9 [dashboard] fix peek parse message error (#4918) a8b57c9 is described below commit a8b57c97c058862bce1c5b9d3435768d2071f09b Author: Yi Tang AuthorDate: Wed Aug 14 09:25:50 2019 +0800 [dashboard] fix peek parse message error (#4918) Fixes #4917 ### Motivation dashboard peek message api raise exception when format response cause treat all message as a JSON string ### Modifications * format as JSON only if message is a JSON, * otherwise if message is printable, return the original message, * otherwise print hex like command with --hex option. --- dashboard/Dockerfile | 2 +- dashboard/django/stats/templates/stats/peek.html | 5 ++- dashboard/django/stats/views.py | 47 ++-- 3 files changed, 49 insertions(+), 5 deletions(-) diff --git a/dashboard/Dockerfile b/dashboard/Dockerfile index 01c84ea..6af19c4 100644 --- a/dashboard/Dockerfile +++ b/dashboard/Dockerfile @@ -25,7 +25,7 @@ RUN apt-get update RUN apt-get -y install postgresql python sudo nginx supervisor # Python dependencies -RUN pip install uwsgi 'Django<2.0' psycopg2 pytz requests +RUN pip install uwsgi 'Django<2.0' psycopg2 pytz requests hexdump # Postgres configuration COPY conf/postgresql.conf /etc/postgresql/9.6/main/ diff --git a/dashboard/django/stats/templates/stats/peek.html b/dashboard/django/stats/templates/stats/peek.html index ebe4821..17d66b0 100644 --- a/dashboard/django/stats/templates/stats/peek.html +++ b/dashboard/django/stats/templates/stats/peek.html @@ -19,4 +19,7 @@ --> -{{ message_body }} \ No newline at end of file + +{{ message_type }} +{{ message_body }} + diff --git a/dashboard/django/stats/views.py b/dashboard/django/stats/views.py index b49994b..d0a7af4 100644 --- a/dashboard/django/stats/views.py +++ b/dashboard/django/stats/views.py @@ -18,12 +18,15 @@ # import logging +import struct + from django.shortcuts import render, get_object_or_404, redirect from django.template import loader from django.urls import reverse from django.views import generic from django.db.models import Q, IntegerField from dashboard import settings +import hexdump import requests, json, re from django.http import HttpResponseRedirect, HttpResponse @@ -363,13 +366,51 @@ def messages(request, topic_name, subscription_name): 'subtitle' : subscription_name, }) + +def message_skip_meta(message_view): +if not message_view or len(message_view) < 4: +raise ValueError("invalid message") +meta_size = struct.unpack(">I", message_view[:4]) +message_index = 4 + meta_size[0] +if len(message_view) < message_index: +raise ValueError("invalid message") +return message_view[message_index:] + + +def get_message_from_http_response(response): +if response.status_code != 200: +return "ERROR", "status_code=%d" % response.status_code +message_view = memoryview(response.content) +if 'X-Pulsar-num-batch-message' in response.headers: +batch_size = int(response.headers['X-Pulsar-num-batch-message']) +if batch_size == 1: +message_view = message_skip_meta(message_view) +else: +# TODO: can not figure out multi-message batch for now +return "Batch(size=%d)" % batch_size, "" + +try: +text = str(message_view, + encoding=response.encoding or response.apparent_encoding, + errors='replace') +if not text.isprintable(): +return "Hex", hexdump.hexdump(message_view, result='return') +except (LookupError, TypeError): +return "Hex", hexdump.hexdump(message_view, result='return') +try: +return "JSON", json.dumps(json.loads(text), + ensure_ascii=False, indent=4) +except json.JSONDecodeError: +return "Text", text + + def peek(request, topic_name, subscription_name, message_number): url = settings.SERVICE_URL + '/admin/v2/' + topic_name + '/subscription/' + subscription_name + '/position/' + message_number response = requests.get(url) -message = response.text -message = message[message.index('{'):] +message_type, message = get_message_from_http_response(response) context = { -'message_body' : json.dumps(json.loads(message), indent=4), +'message_type': message_type, +'message_body': message, } return render(request, 'stats/peek.html', context)
[GitHub] [pulsar] sijie closed issue #4917: [dashboard] peek parse response raise exception
sijie closed issue #4917: [dashboard] peek parse response raise exception URL: https://github.com/apache/pulsar/issues/4917 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: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [pulsar] sijie merged pull request #4918: [dashboard] fix peek parse message error
sijie merged pull request #4918: [dashboard] fix peek parse message error URL: https://github.com/apache/pulsar/pull/4918 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: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [pulsar] sijie commented on a change in pull request #4911: expose getLastMessageId method in ConsumerImpl
sijie commented on a change in pull request #4911: expose getLastMessageId method in ConsumerImpl URL: https://github.com/apache/pulsar/pull/4911#discussion_r313676335 ## File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiMessageIdImpl.java ## @@ -0,0 +1,111 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.client.impl; + +import java.util.Map; +import java.util.Map.Entry; +import java.util.Objects; +import lombok.Getter; +import org.apache.pulsar.client.api.MessageId; +import sun.reflect.generics.reflectiveObjects.NotImplementedException; + +/** + * A MessageId implementation that contains a map of . + * This is useful when MessageId is need for partition/multi-topics/pattern consumer. + * e.g. seek(), ackCumulative(), getLastMessageId(). + */ +public class MultiMessageIdImpl implements MessageId { +@Getter +private Map map; + +MultiMessageIdImpl(Map map) { +this.map = map; +} + +// TODO: Add support for Serialization and Deserialization +// https://github.com/apache/pulsar/issues/4940 +@Override +public byte[] toByteArray() { +throw new NotImplementedException(); +} + +@Override +public int hashCode() { +return Objects.hash(map); +} + +// If all messageId in map are same size, and all bigger/smaller than the other, return valid value. +@Override +public int compareTo(MessageId o) { +if (!(o instanceof MultiMessageIdImpl)) { +throw new IllegalArgumentException( +"expected MultiMessageIdImpl object. Got instance of " + o.getClass().getName()); +} + +MultiMessageIdImpl other = (MultiMessageIdImpl) o; +Map otherMap = other.getMap(); + +if ((map == null || map.isEmpty()) && (otherMap == null || otherMap.isEmpty())) { +return 0; +} + +if (otherMap == null || map == null || otherMap.size() != map.size()) { +throw new IllegalArgumentException("Current size and other size not equals"); +} + +int result = 0; +for (Entry entry : map.entrySet()) { +MessageId otherMessage = otherMap.get(entry.getKey()); +if (otherMessage == null) { +throw new IllegalArgumentException( +"Other MessageId not have topic " + entry.getKey()); +} + +int currentResult = entry.getValue().compareTo(otherMessage); +if (result == 0) { +result = currentResult; Review comment: ok 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: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [pulsar-client-go] wolfstudy opened a new issue #53: Fatal error: concurrent map writes
wolfstudy opened a new issue #53: Fatal error: concurrent map writes URL: https://github.com/apache/pulsar-client-go/issues/53 Expected behavior The program works fine when we create multiple partitions. Actual behavior ``` fatal error: concurrent map writes goroutine 119 [running]: runtime.throw(0x4501a47, 0x15) /usr/local/Cellar/go/1.12.5/libexec/src/runtime/panic.go:617 +0x72 fp=0xc0003dfe88 sp=0xc0003dfe58 pc=0x402fbf2 runtime.mapassign_fast64(0x4478040, 0xc000277a40, 0x34, 0x0) /usr/local/Cellar/go/1.12.5/libexec/src/runtime/map_fast64.go:101 +0x35f fp=0xc0003dfec8 sp=0xc0003dfe88 pc=0x4013cdf github.com/apache/pulsar-client-go/pulsar/internal.(*connection).run(0xc00016e2a0) /Users/wolf4j/github.com/apache/pulsar-client-go/pulsar/internal/connection.go:261 +0x19d fp=0xc0003dffb8 sp=0xc0003dfec8 pc=0x430c55d github.com/apache/pulsar-client-go/pulsar/internal.(*connection).start.func1(0xc00016e2a0) /Users/wolf4j/github.com/apache/pulsar-client-go/pulsar/internal/connection.go:146 +0x59 fp=0xc0003dffd8 sp=0xc0003dffb8 pc=0x4311b09 runtime.goexit() /usr/local/Cellar/go/1.12.5/libexec/src/runtime/asm_amd64.s:1337 +0x1 fp=0xc0003dffe0 sp=0xc0003dffd8 pc=0x405e331 created by github.com/apache/pulsar-client-go/pulsar/internal.(*connection).start /Users/wolf4j/github.com/apache/pulsar-client-go/pulsar/internal/connection.go:143 +0x3f goroutine 1 [chan receive]: testing.(*T).Run(0xc000466000, 0x45075c9, 0x21, 0x4516718, 0x407cb01) /usr/local/Cellar/go/1.12.5/libexec/src/testing/testing.go:917 +0x381 testing.runTests.func1(0xc00015e000) /usr/local/Cellar/go/1.12.5/libexec/src/testing/testing.go:1157 +0x78 testing.tRunner(0xc00015e000, 0xce7e30) /usr/local/Cellar/go/1.12.5/libexec/src/testing/testing.go:865 +0xc0 testing.runTests(0xccc480, 0x48a7d00, 0x23, 0x23, 0x0) /usr/local/Cellar/go/1.12.5/libexec/src/testing/testing.go:1155 +0x2a9 testing.(*M).Run(0xc00014, 0x0) /usr/local/Cellar/go/1.12.5/libexec/src/testing/testing.go:1072 +0x162 main.main() _testmain.go:110 +0x13e goroutine 81 [chan receive]: github.com/apache/pulsar-client-go/pulsar.newProducer(0xc000222780, 0xc000141480, 0xc0003803f0, 0xc000466100, 0xc000315ef0) /Users/wolf4j/github.com/apache/pulsar-client-go/pulsar/impl_producer.go:92 +0x1e4 github.com/apache/pulsar-client-go/pulsar.(*client).CreateProducer(0xc000222780, 0x450cf17, 0x2d, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, ...) /Users/wolf4j/github.com/apache/pulsar-client-go/pulsar/impl_client.go:91 +0x7a github.com/apache/pulsar-client-go/pulsar.TestPartitionTopicsConsumerPubSub(0xc000466000) /Users/wolf4j/github.com/apache/pulsar-client-go/pulsar/consumer_test.go:367 +0x277 testing.tRunner(0xc000466000, 0x4516718) /usr/local/Cellar/go/1.12.5/libexec/src/testing/testing.go:865 +0xc0 created by testing.(*T).Run /usr/local/Cellar/go/1.12.5/libexec/src/testing/testing.go:916 +0x35a ``` Steps to reproduce How can we reproduce the issue 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: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [pulsar-client-go] wolfstudy merged pull request #43: Support partition consumer receive async and fix batch logic
wolfstudy merged pull request #43: Support partition consumer receive async and fix batch logic URL: https://github.com/apache/pulsar-client-go/pull/43 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: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [pulsar-client-go] wolfstudy closed issue #36: Support partition-consumer receive async logic
wolfstudy closed issue #36: Support partition-consumer receive async logic URL: https://github.com/apache/pulsar-client-go/issues/36 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: us...@infra.apache.org With regards, Apache Git Services
[pulsar-client-go] branch master updated: Support partition consumer receive async and fix batch logic (#43)
This is an automated email from the ASF dual-hosted git repository. rxl pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/pulsar-client-go.git The following commit(s) were added to refs/heads/master by this push: new 448387d Support partition consumer receive async and fix batch logic (#43) 448387d is described below commit 448387d738a2f3af4c8232daa4fac9576d252617 Author: 冉小龙 AuthorDate: Wed Aug 14 11:00:00 2019 +0800 Support partition consumer receive async and fix batch logic (#43) Signed-off-by: xiaolong.ran ranxiaolong...@gmail.com * Support batch logic for project * add unit test case of event time * add some unit tests case for producer * fix error result type * add unit test case of producer flush * add receiver queue size test logic * support partition consumer receive async * add unit test case of ack timeout * Fix consumer receiving message out of order --- pulsar/consumer.go| 3 + pulsar/consumer_test.go | 301 +- pulsar/error.go | 2 +- pulsar/impl_consumer.go | 55 +-- pulsar/impl_partition_consumer.go | 217 --- pulsar/impl_partition_producer.go | 21 ++- pulsar/internal/commands.go | 82 +++ pulsar/internal/connection.go | 54 +++ pulsar/producer_test.go | 260 +++- pulsar/unackedMsgTracker.go | 20 +-- util/util.go | 24 ++- util/util_test.go | 21 ++- 12 files changed, 870 insertions(+), 190 deletions(-) diff --git a/pulsar/consumer.go b/pulsar/consumer.go index de190e0..c259cd6 100644 --- a/pulsar/consumer.go +++ b/pulsar/consumer.go @@ -142,6 +142,9 @@ type Consumer interface { // ReceiveAsync appends the message to the msgs channel asynchronously. ReceiveAsync(ctx context.Context, msgs chan<- ConsumerMessage) error + // ReceiveAsyncWithCallback returns a callback containing the message and error objects + ReceiveAsyncWithCallback(ctx context.Context, callback func(msg Message, err error)) + // Ack the consumption of a single message Ack(Message) error diff --git a/pulsar/consumer_test.go b/pulsar/consumer_test.go index 39646d3..6fe86cd 100644 --- a/pulsar/consumer_test.go +++ b/pulsar/consumer_test.go @@ -1,4 +1,3 @@ -// // Licensed to the Apache Software Foundation (ASF) under one // or more contributor license agreements. See the NOTICE file // distributed with this work for additional information @@ -15,7 +14,6 @@ // KIND, either express or implied. See the License for the // specific language governing permissions and limitations // under the License. -// package pulsar @@ -124,6 +122,67 @@ func TestConsumerConnectError(t *testing.T) { assert.Equal(t, err.Error(), "connection error") } +func TestBatchMessageReceive(t *testing.T) { + client, err := NewClient(ClientOptions{ + URL: lookupURL, + }) + + assert.Nil(t, err) + defer client.Close() + + topicName := "persistent://public/default/receive-batch" + subName := "subscription-name" + prefix := "msg-batch-" + ctx := context.Background() + + // Enable batching on producer side + batchSize, numOfMessages := 2, 100 + + // create producer + producer, err := client.CreateProducer(ProducerOptions{ + Topic: topicName, + BatchingMaxMessages: uint(batchSize), + DisableBatching: false, + BlockIfQueueFull:true, + }) + assert.Nil(t, err) + assert.Equal(t, topicName, producer.Topic()) + defer producer.Close() + + consumer, err := client.Subscribe(ConsumerOptions{ + Topic:topicName, + SubscriptionName: subName, + }) + assert.Equal(t, topicName, consumer.Topic()) + count := 0 + + for i := 0; i < numOfMessages; i++ { + messageContent := prefix + fmt.Sprintf("%d", i) + msg := &ProducerMessage{ + Payload: []byte(messageContent), + } + err := producer.Send(ctx, msg) + assert.Nil(t, err) + } + + for i := 0; i < numOfMessages; i++ { + msg, err := consumer.Receive(ctx) + assert.Nil(t, err) + err = consumer.Ack(msg) + assert.Nil(t, err) + count++ + } + + // check strategically + for i := 0; i < 3; i++ { + if count == numOfMessages { + break + } + time.Sleep(time.Second) + } + assert.Equal(t, count, numOfMessages) +} + func TestConsumerWithInvalidConf(t *testing.T) { client, err := NewClient(C
[GitHub] [pulsar-client-go] wolfstudy opened a new pull request #54: [Issue:53]v
wolfstudy opened a new pull request #54: [Issue:53]v URL: https://github.com/apache/pulsar-client-go/pull/54 Signed-off-by: xiaolong.ran Fixes #53 ### Motivation Fix concurrent map write and add test case for this. 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: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [pulsar] kimcs commented on issue #4928: [java-client] ConsumerBuilder should have option for defaultSubscriptionInitialPosition
kimcs commented on issue #4928: [java-client] ConsumerBuilder should have option for defaultSubscriptionInitialPosition URL: https://github.com/apache/pulsar/issues/4928#issuecomment-521124404 I have changed my design and now rely on the pulsar client application to track positions rather than using subscriptions and acknowledgements, so I no longer need this feature for what I'm working on. Still I think this option is useful, as it allows using Pulsar subscriptions to be master of stream progress with at-least-once semantics. 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: us...@infra.apache.org With regards, Apache Git Services