merlimat closed pull request #3003: Refactored C++ tests
URL: https://github.com/apache/pulsar/pull/3003
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/bin/pulsar-perf b/bin/pulsar-perf
index 7a250b7f4c..7c32843e0c 100755
--- a/bin/pulsar-perf
+++ b/bin/pulsar-perf
@@ -21,7 +21,7 @@
BINDIR=$(dirname "$0")
PULSAR_HOME=`cd $BINDIR/..;pwd`
-DEFAULT_CLIENT_CONF=$PULSAR_HOME/conf/client.conf
+DEFAULT_CLIENT_CONF=${PULSAR_CLIENT_CONF:-"$PULSAR_HOME/conf/client.conf"}
DEFAULT_LOG_CONF=$PULSAR_HOME/conf/log4j2.yaml
if [ -f "$PULSAR_HOME/conf/pulsar_env.sh" ]
diff --git a/pulsar-client-cpp/pulsar-test-service-start.sh
b/pulsar-client-cpp/pulsar-test-service-start.sh
new file mode 100755
index 0000000000..914354a3b6
--- /dev/null
+++ b/pulsar-client-cpp/pulsar-test-service-start.sh
@@ -0,0 +1,98 @@
+#!/bin/bash
+#
+# 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.
+#
+
+set -e
+
+SRC_DIR=$(git rev-parse --show-toplevel)
+cd $SRC_DIR
+
+if [ -f /.dockerenv ]; then
+ # When running tests inside docker. Unpack the pulsar tgz
+ # because otherwise the classpath might not be correct
+ # in picking up the jars from local maven repo
+ export PULSAR_DIR=/tmp/pulsar-test-dist
+ rm -rf $PULSAR_DIR
+ mkdir $PULSAR_DIR
+ TGZ=$(ls -1 $SRC_DIR/distribution/server/target/apache-pulsar*bin.tar.gz |
head -1)
+ tar xfz $TGZ -C $PULSAR_DIR --strip-components 1
+else
+ export PULSAR_DIR=$SRC_DIR
+fi
+
+DATA_DIR=/tmp/pulsar-test-data
+rm -rf $DATA_DIR
+mkdir -p $DATA_DIR
+
+# Copy TLS test certificates
+mkdir -p $DATA_DIR/certs
+cp $SRC_DIR/pulsar-broker/src/test/resources/authentication/tls/*.pem
$DATA_DIR/certs
+
+export
PULSAR_STANDALONE_CONF=$SRC_DIR/pulsar-client-cpp/test-conf/standalone-ssl.conf
+$PULSAR_DIR/bin/pulsar-daemon start standalone \
+ --no-functions-worker --no-stream-storage \
+ --zookeeper-dir $DATA_DIR/zookeeper \
+ --bookkeeper-dir $DATA_DIR/bookkeeper
+
+echo "-- Wait for Pulsar service to be ready"
+until curl http://localhost:8080/metrics > /dev/null 2>&1 ; do sleep 1; done
+
+echo "-- Pulsar service is ready -- Configure permissions"
+
+export PULSAR_CLIENT_CONF=$SRC_DIR/pulsar-client-cpp/test-conf/client-ssl.conf
+
+# Create "standalone" cluster
+$PULSAR_DIR/bin/pulsar-admin clusters create \
+ standalone \
+ --url http://localhost:8080/ \
+ --url-secure https://localhost:8443/ \
+ --broker-url pulsar://localhost:6650/ \
+ --broker-url-secure pulsar+ssl://localhost:6651/
+
+# Create "public" tenant
+$PULSAR_DIR/bin/pulsar-admin tenants create public -r "anonymous" -c
"standalone"
+
+# Create "public/default" with no auth required
+$PULSAR_DIR/bin/pulsar-admin namespaces create public/default \
+ --clusters standalone
+$PULSAR_DIR/bin/pulsar-admin namespaces grant-permission public/default \
+ --actions produce,consume \
+ --role "anonymous"
+
+# Create "public/default-2" with no auth required
+$PULSAR_DIR/bin/pulsar-admin namespaces create public/default-2 \
+ --clusters standalone
+$PULSAR_DIR/bin/pulsar-admin namespaces grant-permission public/default-2 \
+ --actions produce,consume \
+ --role "anonymous"
+
+# Create "public/default-3" with no auth required
+$PULSAR_DIR/bin/pulsar-admin namespaces create public/default-3 \
+ --clusters standalone
+$PULSAR_DIR/bin/pulsar-admin namespaces grant-permission public/default-3 \
+ --actions produce,consume \
+ --role "anonymous"
+
+# Create "private" tenant
+$PULSAR_DIR/bin/pulsar-admin tenants create private -r "" -c "standalone"
+
+# Create "private/auth" with required authentication
+$PULSAR_DIR/bin/pulsar-admin namespaces create private/auth --clusters
standalone
+
+echo "-- Ready to start tests"
diff --git a/pulsar-client-cpp/pulsar-test-service-stop.sh
b/pulsar-client-cpp/pulsar-test-service-stop.sh
new file mode 100755
index 0000000000..ff67a61c30
--- /dev/null
+++ b/pulsar-client-cpp/pulsar-test-service-stop.sh
@@ -0,0 +1,26 @@
+#!/bin/bash
+#
+# 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.
+#
+
+set -e
+
+ROOT_DIR=$(git rev-parse --show-toplevel)
+cd $ROOT_DIR
+
+bin/pulsar-daemon stop standalone
diff --git a/pulsar-client-cpp/python/pulsar_test.py
b/pulsar-client-cpp/python/pulsar_test.py
index 86bad9a86d..e9ecbc1b00 100755
--- a/pulsar-client-cpp/python/pulsar_test.py
+++ b/pulsar-client-cpp/python/pulsar_test.py
@@ -41,20 +41,32 @@ def doHttpPost(url, data):
req.add_header('Content-Type', 'application/json')
urlopen(req)
-import urllib2
+
def doHttpPut(url, data):
- opener = urllib2.build_opener(urllib2.HTTPHandler)
- request = urllib2.Request(url, data=data.encode())
- request.add_header('Content-Type', 'application/json')
- request.get_method = lambda: 'PUT'
- opener.open(request)
+ try:
+ req = Request(url, data.encode())
+ req.add_header('Content-Type', 'application/json')
+ req.get_method = lambda: 'PUT'
+ urlopen(req)
+ except Exception as ex:
+ # ignore conflicts exception to have test idempotency
+ if '409' in str(ex):
+ pass
+ else:
+ raise ex
+
+
+def doHttpGet(url):
+ req = Request(url)
+ req.add_header('Accept', 'application/json')
+ return urlopen(req).read()
class PulsarTest(TestCase):
- serviceUrl = 'pulsar://localhost:8885'
- adminUrl = 'http://localhost:8765'
+ serviceUrl = 'pulsar://localhost:6650'
+ adminUrl = 'http://localhost:8080'
- serviceUrlTls = 'pulsar+ssl://localhost:9886'
+ serviceUrlTls = 'pulsar+ssl://localhost:6651'
def test_producer_config(self):
conf = ProducerConfiguration()
@@ -80,23 +92,23 @@ def test_consumer_config(self):
def test_simple_producer(self):
client = Client(self.serviceUrl)
- producer =
client.create_producer('persistent://sample/standalone/ns/my-python-topic')
- producer.send('hello')
+ producer = client.create_producer('my-python-topic')
+ producer.send(b'hello')
producer.close()
client.close()
def test_producer_send_async(self):
client = Client(self.serviceUrl)
- producer =
client.create_producer('persistent://sample/standalone/ns/my-python-topic')
+ producer = client.create_producer('my-python-topic')
sent_messages = []
def send_callback(producer, msg):
sent_messages.append(msg)
- producer.send_async('hello', send_callback)
- producer.send_async('hello', send_callback)
- producer.send_async('hello', send_callback)
+ producer.send_async(b'hello', send_callback)
+ producer.send_async(b'hello', send_callback)
+ producer.send_async(b'hello', send_callback)
i = 0
while len(sent_messages) < 3 and i < 100:
@@ -107,11 +119,11 @@ def send_callback(producer, msg):
def test_producer_consumer(self):
client = Client(self.serviceUrl)
- consumer =
client.subscribe('persistent://sample/standalone/ns/my-python-topic-producer-consumer',
+ consumer = client.subscribe('my-python-topic-producer-consumer',
'my-sub',
consumer_type=ConsumerType.Shared)
- producer =
client.create_producer('persistent://sample/standalone/ns/my-python-topic-producer-consumer')
- producer.send('hello')
+ producer = client.create_producer('my-python-topic-producer-consumer')
+ producer.send(b'hello')
msg = consumer.receive(1000)
self.assertTrue(msg)
@@ -134,11 +146,11 @@ def test_tls_auth(self):
tls_allow_insecure_connection=False,
authentication=AuthenticationTLS(certs_dir +
'client-cert.pem', certs_dir + 'client-key.pem'))
- consumer =
client.subscribe('persistent://property/cluster/namespace/my-python-topic-producer-consumer',
+ consumer = client.subscribe('my-python-topic-producer-consumer',
'my-sub',
consumer_type=ConsumerType.Shared)
- producer =
client.create_producer('persistent://property/cluster/namespace/my-python-topic-producer-consumer')
- producer.send('hello')
+ producer = client.create_producer('my-python-topic-producer-consumer')
+ producer.send(b'hello')
msg = consumer.receive(1000)
self.assertTrue(msg)
@@ -164,11 +176,11 @@ def test_tls_auth2(self):
tls_allow_insecure_connection=False,
authentication=Authentication(authPlugin, authParams))
- consumer =
client.subscribe('persistent://property/cluster/namespace/my-python-topic-producer-consumer',
+ consumer = client.subscribe('my-python-topic-producer-consumer',
'my-sub',
consumer_type=ConsumerType.Shared)
- producer =
client.create_producer('persistent://property/cluster/namespace/my-python-topic-producer-consumer')
- producer.send('hello')
+ producer = client.create_producer('my-python-topic-producer-consumer')
+ producer.send(b'hello')
msg = consumer.receive(1000)
self.assertTrue(msg)
@@ -194,11 +206,11 @@ def test_tls_auth3(self):
tls_allow_insecure_connection=False,
authentication=Authentication(authPlugin, authParams))
- consumer =
client.subscribe('persistent://property/cluster/namespace/my-python-topic-producer-consumer',
+ consumer = client.subscribe('my-python-topic-producer-consumer',
'my-sub',
consumer_type=ConsumerType.Shared)
- producer =
client.create_producer('persistent://property/cluster/namespace/my-python-topic-producer-consumer')
- producer.send('hello')
+ producer = client.create_producer('my-python-topic-producer-consumer')
+ producer.send(b'hello')
msg = consumer.receive(1000)
self.assertTrue(msg)
@@ -223,7 +235,7 @@ def test_auth_junk_params(self):
tls_allow_insecure_connection=False,
authentication=Authentication(authPlugin, authParams))
try:
-
client.subscribe('persistent://property/cluster/namespace/my-python-topic-producer-consumer',
+ client.subscribe('my-python-topic-producer-consumer',
'my-sub',
consumer_type=ConsumerType.Shared)
except:
@@ -239,14 +251,14 @@ def listener(consumer, msg):
received_messages.append(msg)
consumer.acknowledge(msg)
-
client.subscribe('persistent://sample/standalone/ns/my-python-topic-listener',
+ client.subscribe('my-python-topic-listener',
'my-sub',
consumer_type=ConsumerType.Exclusive,
message_listener=listener)
- producer =
client.create_producer('persistent://sample/standalone/ns/my-python-topic-listener')
- producer.send('hello-1')
- producer.send('hello-2')
- producer.send('hello-3')
+ producer = client.create_producer('my-python-topic-listener')
+ producer.send(b'hello-1')
+ producer.send(b'hello-2')
+ producer.send(b'hello-3')
time.sleep(0.1)
self.assertEqual(len(received_messages), 3)
@@ -257,11 +269,11 @@ def listener(consumer, msg):
def test_reader_simple(self):
client = Client(self.serviceUrl)
- reader =
client.create_reader('persistent://sample/standalone/ns/my-python-topic-reader-simple',
+ reader = client.create_reader('my-python-topic-reader-simple',
MessageId.earliest)
- producer =
client.create_producer('persistent://sample/standalone/ns/my-python-topic-reader-simple')
- producer.send('hello')
+ producer = client.create_producer('my-python-topic-reader-simple')
+ producer.send(b'hello')
msg = reader.read_next()
self.assertTrue(msg)
@@ -278,16 +290,16 @@ def test_reader_simple(self):
def test_reader_on_last_message(self):
client = Client(self.serviceUrl)
- producer =
client.create_producer('persistent://sample/standalone/ns/my-python-topic-reader-on-last-message')
+ producer =
client.create_producer('my-python-topic-reader-on-last-message')
for i in range(10):
- producer.send('hello-%d' % i)
+ producer.send(b'hello-%d' % i)
- reader =
client.create_reader('persistent://sample/standalone/ns/my-python-topic-reader-on-last-message',
+ reader = client.create_reader('my-python-topic-reader-on-last-message',
MessageId.latest)
for i in range(10, 20):
- producer.send('hello-%d' % i)
+ producer.send(b'hello-%d' % i)
for i in range(10, 20):
msg = reader.read_next()
@@ -300,13 +312,13 @@ def test_reader_on_last_message(self):
def test_reader_on_specific_message(self):
client = Client(self.serviceUrl)
producer = client.create_producer(
-
'persistent://sample/standalone/ns/my-python-topic-reader-on-specific-message')
+ 'my-python-topic-reader-on-specific-message')
for i in range(10):
- producer.send('hello-%d' % i)
+ producer.send(b'hello-%d' % i)
reader1 = client.create_reader(
-
'persistent://sample/standalone/ns/my-python-topic-reader-on-specific-message',
+ 'my-python-topic-reader-on-specific-message',
MessageId.earliest)
for i in range(5):
@@ -314,7 +326,7 @@ def test_reader_on_specific_message(self):
last_msg_id = msg.message_id()
reader2 = client.create_reader(
-
'persistent://sample/standalone/ns/my-python-topic-reader-on-specific-message',
+ 'my-python-topic-reader-on-specific-message',
last_msg_id)
for i in range(5, 10):
@@ -329,18 +341,18 @@ def test_reader_on_specific_message(self):
def test_reader_on_specific_message_with_batches(self):
client = Client(self.serviceUrl)
producer = client.create_producer(
-
'persistent://sample/standalone/ns/my-python-topic-reader-on-specific-message-with-batches',
+ 'my-python-topic-reader-on-specific-message-with-batches',
batching_enabled=True,
batching_max_publish_delay_ms=1000)
for i in range(10):
- producer.send_async('hello-%d' % i, None)
+ producer.send_async(b'hello-%d' % i, None)
# Send one sync message to make sure everything was published
- producer.send('hello-10')
+ producer.send(b'hello-10')
reader1 = client.create_reader(
-
'persistent://sample/standalone/ns/my-python-topic-reader-on-specific-message-with-batches',
+ 'my-python-topic-reader-on-specific-message-with-batches',
MessageId.earliest)
for i in range(5):
@@ -348,7 +360,7 @@ def test_reader_on_specific_message_with_batches(self):
last_msg_id = msg.message_id()
reader2 = client.create_reader(
-
'persistent://sample/standalone/ns/my-python-topic-reader-on-specific-message-with-batches',
+ 'my-python-topic-reader-on-specific-message-with-batches',
last_msg_id)
for i in range(5, 11):
@@ -362,18 +374,18 @@ def test_reader_on_specific_message_with_batches(self):
def test_producer_sequence_after_reconnection(self):
# Enable deduplication on namespace
- doHttpPost(self.adminUrl +
'/admin/namespaces/sample/standalone/ns1/deduplication',
+ doHttpPost(self.adminUrl +
'/admin/v2/namespaces/public/default/deduplication',
'true')
client = Client(self.serviceUrl)
- topic =
'persistent://sample/standalone/ns1/my-python-test-producer-sequence-after-reconnection-'
\
+ topic = 'my-python-test-producer-sequence-after-reconnection-' \
+ str(time.time())
producer = client.create_producer(topic,
producer_name='my-producer-name')
self.assertEqual(producer.last_sequence_id(), -1)
for i in range(10):
- producer.send('hello-%d' % i)
+ producer.send(b'hello-%d' % i)
self.assertEqual(producer.last_sequence_id(), i)
producer.close()
@@ -382,30 +394,33 @@ def test_producer_sequence_after_reconnection(self):
self.assertEqual(producer.last_sequence_id(), 9)
for i in range(10, 20):
- producer.send('hello-%d' % i)
+ producer.send(b'hello-%d' % i)
self.assertEqual(producer.last_sequence_id(), i)
+ doHttpPost(self.adminUrl +
'/admin/v2/namespaces/public/default/deduplication',
+ 'false')
+
def test_producer_deduplication(self):
# Enable deduplication on namespace
- doHttpPost(self.adminUrl +
'/admin/namespaces/sample/standalone/ns1/deduplication',
+ doHttpPost(self.adminUrl +
'/admin/v2/namespaces/public/default/deduplication',
'true')
client = Client(self.serviceUrl)
- topic =
'persistent://sample/standalone/ns1/my-python-test-producer-deduplication-' +
str(time.time())
+ topic = 'my-python-test-producer-deduplication-' + str(time.time())
producer = client.create_producer(topic,
producer_name='my-producer-name')
self.assertEqual(producer.last_sequence_id(), -1)
consumer = client.subscribe(topic, 'my-sub')
- producer.send('hello-0', sequence_id=0)
- producer.send('hello-1', sequence_id=1)
- producer.send('hello-2', sequence_id=2)
+ producer.send(b'hello-0', sequence_id=0)
+ producer.send(b'hello-1', sequence_id=1)
+ producer.send(b'hello-2', sequence_id=2)
self.assertEqual(producer.last_sequence_id(), 2)
# Repeat the messages and verify they're not received by consumer
- producer.send('hello-1', sequence_id=1)
- producer.send('hello-2', sequence_id=2)
+ producer.send(b'hello-1', sequence_id=1)
+ producer.send(b'hello-2', sequence_id=2)
self.assertEqual(producer.last_sequence_id(), 2)
for i in range(3):
@@ -427,8 +442,8 @@ def test_producer_deduplication(self):
self.assertEqual(producer.last_sequence_id(), 2)
# Repeat the messages and verify they're not received by consumer
- producer.send('hello-1', sequence_id=1)
- producer.send('hello-2', sequence_id=2)
+ producer.send(b'hello-1', sequence_id=1)
+ producer.send(b'hello-2', sequence_id=2)
self.assertEqual(producer.last_sequence_id(), 2)
try:
@@ -439,6 +454,9 @@ def test_producer_deduplication(self):
# Exception is expected
pass
+ doHttpPost(self.adminUrl +
'/admin/v2/namespaces/public/default/deduplication',
+ 'false')
+
def test_producer_routing_mode(self):
client = Client(self.serviceUrl)
producer = client.create_producer('my-python-test-producer',
@@ -448,7 +466,7 @@ def test_producer_routing_mode(self):
def test_message_argument_errors(self):
client = Client(self.serviceUrl)
- topic = 'persistent://sample/standalone/ns1/my-python-test-producer'
+ topic = 'my-python-test-producer'
producer = client.create_producer(topic)
content = 'test'.encode('utf-8')
@@ -478,7 +496,7 @@ def test_producer_argument_errors(self):
self._check_value_error(lambda: client.create_producer(None))
- topic = 'persistent://sample/standalone/ns1/my-python-test-producer'
+ topic = 'my-python-test-producer'
self._check_value_error(lambda: client.create_producer(topic,
producer_name=5))
self._check_value_error(lambda: client.create_producer(topic,
initial_sequence_id='test'))
@@ -495,7 +513,7 @@ def test_producer_argument_errors(self):
def test_consumer_argument_errors(self):
client = Client(self.serviceUrl)
- topic = 'persistent://sample/standalone/ns1/my-python-test-producer'
+ topic = 'my-python-test-producer'
sub_name = 'my-sub-name'
self._check_value_error(lambda: client.subscribe(None, sub_name))
@@ -509,7 +527,7 @@ def test_consumer_argument_errors(self):
def test_reader_argument_errors(self):
client = Client(self.serviceUrl)
- topic = 'persistent://sample/standalone/ns1/my-python-test-producer'
+ topic = 'my-python-test-producer'
# This should not raise exception
client.create_reader(topic, MessageId.earliest)
@@ -522,7 +540,7 @@ def test_reader_argument_errors(self):
def test_publish_compact_and_consume(self):
client = Client(self.serviceUrl)
- topic =
'persistent://sample/standalone/ns1/my-python-test_publish_compact_and_consume'
+ topic = 'my-python-test_publish_compact_and_consume'
producer = client.create_producer(topic,
producer_name='my-producer-name', batching_enabled=False)
self.assertEqual(producer.last_sequence_id(), -1)
consumer = client.subscribe(topic, 'my-sub1', is_read_compacted=True)
@@ -530,17 +548,15 @@ def test_publish_compact_and_consume(self):
consumer2 = client.subscribe(topic, 'my-sub2', is_read_compacted=False)
# producer create 2 messages with same key.
- producer.send('hello-0', partition_key='key0')
- producer.send('hello-1', partition_key='key0')
+ producer.send(b'hello-0', partition_key='key0')
+ producer.send(b'hello-1', partition_key='key0')
producer.close()
# issue compact command, and wait success
- url=self.adminUrl +
'/admin/persistent/sample/standalone/ns1/my-python-test_publish_compact_and_consume/compaction'
+ url=self.adminUrl +
'/admin/v2/persistent/public/default/my-python-test_publish_compact_and_consume/compaction'
doHttpPut(url, '')
while True:
- req = urllib2.Request(url)
- response = urllib2.urlopen(req)
- s=response.read()
+ s=doHttpGet(url).decode('utf-8')
if 'RUNNING' in s:
print("Compact still running")
print(s)
@@ -571,15 +587,15 @@ def test_publish_compact_and_consume(self):
def test_reader_has_message_available(self):
# create client, producer, reader
client = Client(self.serviceUrl)
- producer =
client.create_producer('persistent://sample/standalone/ns/my-python-topic-reader-has-message-available')
- reader =
client.create_reader('persistent://sample/standalone/ns/my-python-topic-reader-has-message-available',
+ producer =
client.create_producer('my-python-topic-reader-has-message-available')
+ reader =
client.create_reader('my-python-topic-reader-has-message-available',
MessageId.latest)
# before produce data, expected not has message available
self.assertFalse(reader.has_message_available());
for i in range(10):
- producer.send('hello-%d' % i)
+ producer.send(b'hello-%d' % i)
# produced data, expected has message available
self.assertTrue(reader.has_message_available());
@@ -593,7 +609,7 @@ def test_reader_has_message_available(self):
self.assertFalse(reader.has_message_available());
for i in range(10, 20):
- producer.send('hello-%d' % i)
+ producer.send(b'hello-%d' % i)
# produced data again, expected has message available
self.assertTrue(reader.has_message_available());
@@ -603,13 +619,13 @@ def test_reader_has_message_available(self):
def test_seek(self):
client = Client(self.serviceUrl)
- consumer =
client.subscribe('persistent://sample/standalone/ns/my-python-topic-seek',
+ consumer = client.subscribe('my-python-topic-seek',
'my-sub',
consumer_type=ConsumerType.Shared)
- producer =
client.create_producer('persistent://sample/standalone/ns/my-python-topic-seek')
+ producer = client.create_producer('my-python-topic-seek')
for i in range(100):
- producer.send('hello-%d' % i)
+ producer.send(b'hello-%d' % i)
for i in range(100):
msg = consumer.receive()
@@ -635,7 +651,7 @@ def _v2_topics(self, url):
'my-sub',
consumer_type=ConsumerType.Shared)
producer = client.create_producer('my-v2-topic-producer-consumer')
- producer.send('hello')
+ producer.send(b'hello')
msg = consumer.receive(1000)
self.assertTrue(msg)
@@ -652,14 +668,14 @@ def _v2_topics(self, url):
def test_topics_consumer(self):
client = Client(self.serviceUrl)
- topic1 =
'persistent://sample/standalone/ns/my-python-topics-consumer-1'
- topic2 =
'persistent://sample/standalone/ns/my-python-topics-consumer-2'
- topic3 =
'persistent://sample/standalone/ns/my-python-topics-consumer-3'
+ topic1 = 'persistent://public/default/my-python-topics-consumer-1'
+ topic2 = 'persistent://public/default/my-python-topics-consumer-2'
+ topic3 = 'persistent://public/default/my-python-topics-consumer-3'
topics = [topic1, topic2, topic3]
- url1 = self.adminUrl +
'/admin/persistent/sample/standalone/ns/my-python-topics-consumer-1/partitions'
- url2 = self.adminUrl +
'/admin/persistent/sample/standalone/ns/my-python-topics-consumer-2/partitions'
- url3 = self.adminUrl +
'/admin/persistent/sample/standalone/ns/my-python-topics-consumer-3/partitions'
+ url1 = self.adminUrl +
'/admin/v2/persistent/public/default/my-python-topics-consumer-1/partitions'
+ url2 = self.adminUrl +
'/admin/v2/persistent/public/default/my-python-topics-consumer-2/partitions'
+ url3 = self.adminUrl +
'/admin/v2/persistent/public/default/my-python-topics-consumer-3/partitions'
doHttpPut(url1, '2')
doHttpPut(url2, '3')
@@ -676,13 +692,13 @@ def test_topics_consumer(self):
)
for i in range(100):
- producer1.send('hello-1-%d' % i)
+ producer1.send(b'hello-1-%d' % i)
for i in range(100):
- producer2.send('hello-2-%d' % i)
+ producer2.send(b'hello-2-%d' % i)
for i in range(100):
- producer3.send('hello-3-%d' % i)
+ producer3.send(b'hello-3-%d' % i)
for i in range(300):
@@ -702,15 +718,15 @@ def test_topics_pattern_consumer(self):
import re
client = Client(self.serviceUrl)
- topics_pattern =
'persistent://sample/standalone/ns/my-python-pattern-consumer.*'
+ topics_pattern =
'persistent://public/default/my-python-pattern-consumer.*'
- topic1 =
'persistent://sample/standalone/ns/my-python-pattern-consumer-1'
- topic2 =
'persistent://sample/standalone/ns/my-python-pattern-consumer-2'
- topic3 =
'persistent://sample/standalone/ns/my-python-pattern-consumer-3'
+ topic1 = 'persistent://public/default/my-python-pattern-consumer-1'
+ topic2 = 'persistent://public/default/my-python-pattern-consumer-2'
+ topic3 = 'persistent://public/default/my-python-pattern-consumer-3'
- url1 = self.adminUrl +
'/admin/persistent/sample/standalone/ns/my-python-pattern-consumer-1/partitions'
- url2 = self.adminUrl +
'/admin/persistent/sample/standalone/ns/my-python-pattern-consumer-2/partitions'
- url3 = self.adminUrl +
'/admin/persistent/sample/standalone/ns/my-python-pattern-consumer-3/partitions'
+ url1 = self.adminUrl +
'/admin/v2/persistent/public/default/my-python-pattern-consumer-1/partitions'
+ url2 = self.adminUrl +
'/admin/v2/persistent/public/default/my-python-pattern-consumer-2/partitions'
+ url3 = self.adminUrl +
'/admin/v2/persistent/public/default/my-python-pattern-consumer-3/partitions'
doHttpPut(url1, '2')
doHttpPut(url2, '3')
@@ -731,13 +747,13 @@ def test_topics_pattern_consumer(self):
time.sleep(2)
for i in range(100):
- producer1.send('hello-1-%d' % i)
+ producer1.send(b'hello-1-%d' % i)
for i in range(100):
- producer2.send('hello-2-%d' % i)
+ producer2.send(b'hello-2-%d' % i)
for i in range(100):
- producer3.send('hello-3-%d' % i)
+ producer3.send(b'hello-3-%d' % i)
for i in range(300):
@@ -763,7 +779,7 @@ def test_message_id(self):
def test_get_topics_partitions(self):
client = Client(self.serviceUrl)
topic_partitioned =
'persistent://public/default/test_get_topics_partitions'
- topic_non_partitioned =
'persistent://public/default/test_get_topics_partitions'
+ topic_non_partitioned =
'persistent://public/default/test_get_topics_not-partitioned'
url1 = self.adminUrl +
'/admin/v2/persistent/public/default/test_get_topics_partitions/partitions'
doHttpPut(url1, '3')
diff --git a/pulsar-client-cpp/run-unit-tests.sh
b/pulsar-client-cpp/run-unit-tests.sh
index cffe0023b7..e9caefa0e7 100755
--- a/pulsar-client-cpp/run-unit-tests.sh
+++ b/pulsar-client-cpp/run-unit-tests.sh
@@ -18,44 +18,12 @@
# under the License.
#
-# Start 2 Pulsar standalone instances (one with TLS and one without)
-# and execute the unit tests
+set -e
-rm -rf ./pulsar-dist
-mkdir pulsar-dist
-tar xfz ../distribution/server/target/apache-pulsar*bin.tar.gz -C pulsar-dist
--strip-components 1
+ROOT_DIR=$(git rev-parse --show-toplevel)
+cd $ROOT_DIR/pulsar-client-cpp
-PULSAR_STANDALONE_CONF=$PWD/test-conf/standalone.conf pulsar-dist/bin/pulsar
standalone --no-functions-worker --no-stream-storage > broker.log &
-standalone_pid=$!;
-
-PULSAR_STANDALONE_CONF=$PWD/test-conf/standalone-ssl.conf
pulsar-dist/bin/pulsar standalone \
- --no-functions-worker \
- --no-stream-storage \
- --zookeeper-port 2191 --bookkeeper-port 3191 \
- --zookeeper-dir data2/standalone/zookeeper --bookkeeper-dir \
- data2/standalone/bookkeeper > broker-tls.log &
-auth_pid=$!;
-
-echo "Wait for non-tls standalone up"
-until grep "Created tenant public" broker.log; do sleep 5; done
-
-# create property for test
-PULSAR_CLIENT_CONF=$PWD/test-conf/client.conf pulsar-dist/bin/pulsar-admin
tenants create prop -r "" -c "unit"
-echo "Created tenant 'prop' - $?"
-
-PULSAR_CLIENT_CONF=$PWD/test-conf/client.conf pulsar-dist/bin/pulsar-admin
tenants create property -r "" -c "cluster"
-echo "Created tenant 'property' - $?"
-
-PULSAR_CLIENT_CONF=$PWD/test-conf/client-ssl.conf pulsar-dist/bin/pulsar-admin
clusters create \
- --url http://localhost:9765/ --url-secure https://localhost:9766/ \
- --broker-url pulsar://localhost:9885/ --broker-url-secure
pulsar+ssl://localhost:9886/ \
- cluster
-PULSAR_CLIENT_CONF=$PWD/test-conf/client-ssl.conf pulsar-dist/bin/pulsar-admin
clusters create \
- --url http://localhost:9765/ --url-secure https://localhost:9766/ \
- --broker-url pulsar://localhost:9885/ --broker-url-secure
pulsar+ssl://localhost:9886/ \
- unit
-
-sleep 5
+./pulsar-test-service-start.sh
pushd tests
@@ -94,11 +62,8 @@ if [ $RES -eq 0 ]; then
popd
popd
-
fi
-kill -9 $standalone_pid $auth_pid
-
-rm -rf pulsar-dist
+./pulsar-test-service-stop.sh
exit $RES
diff --git a/pulsar-client-cpp/test-conf/client-ssl.conf
b/pulsar-client-cpp/test-conf/client-ssl.conf
index d4071b4f77..6ca0e5a6d3 100644
--- a/pulsar-client-cpp/test-conf/client-ssl.conf
+++ b/pulsar-client-cpp/test-conf/client-ssl.conf
@@ -18,10 +18,9 @@
#
# Pulsar Client configuration
-webServiceUrl=https://localhost:9766/
-brokerServiceUrl=pulsar+ssl://localhost:9886/
-useTls=true
-tlsAllowInsecureConnection=true
-tlsTrustCertsFilePath=/pulsar/pulsar-broker/src/test/resources/authentication/tls/cacert.pem
+webServiceUrl=https://localhost:8443/
+brokerServiceUrl=pulsar+ssl://localhost:6651/
+tlsAllowInsecureConnection=false
+tlsTrustCertsFilePath=/tmp/pulsar-test-data/certs/cacert.pem
authPlugin=org.apache.pulsar.client.impl.auth.AuthenticationTls
-authParams=tlsCertFile:/pulsar/pulsar-broker/src/test/resources/authentication/tls/client-cert.pem,tlsKeyFile:/pulsar/pulsar-broker/src/test/resources/authentication/tls/client-key.pem
+authParams=tlsCertFile:/tmp/pulsar-test-data/certs/client-cert.pem,tlsKeyFile:/tmp/pulsar-test-data/certs/client-key.pem
diff --git a/pulsar-client-cpp/test-conf/standalone-ssl.conf
b/pulsar-client-cpp/test-conf/standalone-ssl.conf
index bb097d67c5..da9123c4cc 100644
--- a/pulsar-client-cpp/test-conf/standalone-ssl.conf
+++ b/pulsar-client-cpp/test-conf/standalone-ssl.conf
@@ -28,12 +28,12 @@ globalZookeeperServers=
# Configuration Store connection string
configurationStoreServers=
-brokerServicePort=9885
-brokerServicePortTls=9886
+brokerServicePort=6650
+brokerServicePortTls=6651
# Port to use to server HTTP request
-webServicePort=9765
-webServicePortTls=9766
+webServicePort=8080
+webServicePortTls=8443
# Hostname or IP address the service binds on, default is 0.0.0.0.
bindAddress=0.0.0.0
@@ -42,7 +42,7 @@ bindAddress=0.0.0.0
advertisedAddress=localhost
# Name of the cluster to which this broker belongs to
-clusterName=cluster
+clusterName=standalone
# Zookeeper session timeout in milliseconds
zooKeeperSessionTimeoutMillis=30000
@@ -87,10 +87,12 @@ maxUnackedMessagesPerConsumer=50000
# Enable TLS
tlsEnabled=true
-tlsCertificateFilePath=/pulsar/pulsar-broker/src/test/resources/authentication/tls/broker-cert.pem
-tlsKeyFilePath=/pulsar/pulsar-broker/src/test/resources/authentication/tls/broker-key.pem
-tlsTrustCertsFilePath=/pulsar/pulsar-broker/src/test/resources/authentication/tls/cacert.pem
-tlsAllowInsecureConnection=true
+tlsCertificateFilePath=/tmp/pulsar-test-data/certs/broker-cert.pem
+tlsKeyFilePath=/tmp/pulsar-test-data/certs/broker-key.pem
+tlsTrustCertsFilePath=/tmp/pulsar-test-data/certs/cacert.pem
+tlsAllowInsecureConnection=false
+
+anonymousUserRole=anonymous
# Enable authentication
authenticationEnabled=true
diff --git a/pulsar-client-cpp/tests/AuthPluginTest.cc
b/pulsar-client-cpp/tests/AuthPluginTest.cc
index 295e624b4c..eef541cf18 100644
--- a/pulsar-client-cpp/tests/AuthPluginTest.cc
+++ b/pulsar-client-cpp/tests/AuthPluginTest.cc
@@ -32,7 +32,14 @@ DECLARE_LOG_OBJECT()
using namespace pulsar;
int globalTestTlsMessagesCounter = 0;
-static std::string lookupUrlTls = "pulsar+ssl://localhost:9886";
+static const std::string serviceUrlTls = "pulsar+ssl://localhost:6651";
+static const std::string serviceUrlHttps = "https://localhost:8443";;
+
+static const std::string caPath =
"../../pulsar-broker/src/test/resources/authentication/tls/cacert.pem";
+static const std::string clientPublicKeyPath =
+
"../../pulsar-broker/src/test/resources/authentication/tls/client-cert.pem";
+static const std::string clientPrivateKeyPath =
+ "../../pulsar-broker/src/test/resources/authentication/tls/client-key.pem";
static void sendCallBackTls(Result r, const Message& msg) {
ASSERT_EQ(r, ResultOk);
@@ -44,12 +51,9 @@ static void sendCallBackTls(Result r, const Message& msg) {
TEST(AuthPluginTest, testTls) {
ClientConfiguration config = ClientConfiguration();
- config.setUseTls(true);
-
config.setTlsTrustCertsFilePath("../../pulsar-broker/src/test/resources/authentication/tls/cacert.pem");
+ config.setTlsTrustCertsFilePath(caPath);
config.setTlsAllowInsecureConnection(false);
- AuthenticationPtr auth =
-
pulsar::AuthTls::create("../../pulsar-broker/src/test/resources/authentication/tls/client-cert.pem",
-
"../../pulsar-broker/src/test/resources/authentication/tls/client-key.pem");
+ AuthenticationPtr auth = pulsar::AuthTls::create(clientPublicKeyPath,
clientPrivateKeyPath);
ASSERT_TRUE(auth != NULL);
ASSERT_EQ(auth->getAuthMethodName(), "tls");
@@ -61,9 +65,9 @@ TEST(AuthPluginTest, testTls) {
ASSERT_EQ(auth.use_count(), 1);
config.setAuth(auth);
- Client client(lookupUrlTls, config);
+ Client client(serviceUrlTls, config);
- std::string topicName = "persistent://property/cluster/namespace/test-tls";
+ std::string topicName = "persistent://private/auth/test-tls";
std::string subName = "subscription-name";
int numOfMessages = 10;
@@ -122,16 +126,13 @@ TEST(AuthPluginTest, testTls) {
TEST(AuthPluginTest, testTlsDetectPulsarSsl) {
ClientConfiguration config = ClientConfiguration();
-
config.setTlsTrustCertsFilePath("../../pulsar-broker/src/test/resources/authentication/tls/cacert.pem");
+ config.setTlsTrustCertsFilePath(caPath);
config.setTlsAllowInsecureConnection(false);
- AuthenticationPtr auth =
-
pulsar::AuthTls::create("../../pulsar-broker/src/test/resources/authentication/tls/client-cert.pem",
-
"../../pulsar-broker/src/test/resources/authentication/tls/client-key.pem");
- config.setAuth(auth);
+ config.setAuth(pulsar::AuthTls::create(clientPublicKeyPath,
clientPrivateKeyPath));
- Client client("pulsar+ssl://localhost:9886", config);
+ Client client(serviceUrlTls, config);
- std::string topicName =
"persistent://property/cluster/namespace/test-tls-detect";
+ std::string topicName = "persistent://private/auth/test-tls-detect";
Producer producer;
Promise<Result, Producer> producerPromise;
@@ -144,16 +145,13 @@ TEST(AuthPluginTest, testTlsDetectPulsarSsl) {
TEST(AuthPluginTest, testTlsDetectHttps) {
ClientConfiguration config = ClientConfiguration();
config.setUseTls(true); // shouldn't be needed soon
-
config.setTlsTrustCertsFilePath("../../pulsar-broker/src/test/resources/authentication/tls/cacert.pem");
+ config.setTlsTrustCertsFilePath(caPath);
config.setTlsAllowInsecureConnection(false);
- AuthenticationPtr auth =
-
pulsar::AuthTls::create("../../pulsar-broker/src/test/resources/authentication/tls/client-cert.pem",
-
"../../pulsar-broker/src/test/resources/authentication/tls/client-key.pem");
- config.setAuth(auth);
+ config.setAuth(pulsar::AuthTls::create(clientPublicKeyPath,
clientPrivateKeyPath));
- Client client("https://localhost:9766";, config);
+ Client client(serviceUrlHttps, config);
- std::string topicName =
"persistent://property/cluster/namespace/test-tls-detect-https";
+ std::string topicName = "persistent://private/auth/test-tls-detect-https";
Producer producer;
Promise<Result, Producer> producerPromise;
@@ -165,12 +163,17 @@ TEST(AuthPluginTest, testTlsDetectHttps) {
namespace testAthenz {
std::string principalToken;
-void mockZTS() {
+void mockZTS(int port) {
+ LOG_INFO("-- MockZTS started");
boost::asio::io_service io;
boost::asio::ip::tcp::iostream stream;
boost::asio::ip::tcp::acceptor acceptor(io,
-
boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), 9999));
+
boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), port));
+
+ LOG_INFO("-- MockZTS waiting for connnection");
acceptor.accept(*stream.rdbuf());
+ LOG_INFO("-- MockZTS got connection");
+
std::string headerLine;
while (getline(stream, headerLine)) {
std::vector<std::string> kv;
@@ -190,19 +193,24 @@ void mockZTS() {
break;
}
}
+
+ LOG_INFO("-- MockZTS exiting");
}
} // namespace testAthenz
TEST(AuthPluginTest, testAthenz) {
- boost::thread zts(&testAthenz::mockZTS);
+ boost::thread zts(boost::bind(&testAthenz::mockZTS, 9999));
pulsar::AuthenticationDataPtr data;
std::string params = R"({
"tenantDomain": "pulsar.test.tenant",
"tenantService": "service",
"providerDomain": "pulsar.test.provider",
- "privateKey":
"file:../../pulsar-broker/src/test/resources/authentication/tls/client-key.pem",
+ "privateKey": "file:)" +
+ clientPrivateKeyPath + R"(",
"ztsUrl": "http://localhost:9999";
})";
+
+ LOG_INFO("PARAMS: " << params);
pulsar::AuthenticationPtr auth = pulsar::AuthAthenz::create(params);
ASSERT_EQ(auth->getAuthMethodName(), "athenz");
ASSERT_EQ(auth->getAuthData(data), pulsar::ResultOk);
@@ -237,23 +245,21 @@ TEST(AuthPluginTest, testDisable) {
TEST(AuthPluginTest, testAuthFactoryTls) {
pulsar::AuthenticationDataPtr data;
- std::string tlsCertFile =
"../../pulsar-broker/src/test/resources/authentication/tls/client-cert.pem";
- std::string tlsKeyFile =
"../../pulsar-broker/src/test/resources/authentication/tls/client-key.pem";
- AuthenticationPtr auth =
- pulsar::AuthFactory::create("tls", "tlsCertFile:" + tlsCertFile +
",tlsKeyFile:" + tlsKeyFile);
+ AuthenticationPtr auth = pulsar::AuthFactory::create(
+ "tls", "tlsCertFile:" + clientPublicKeyPath + ",tlsKeyFile:" +
clientPrivateKeyPath);
ASSERT_EQ(auth->getAuthMethodName(), "tls");
ASSERT_EQ(auth->getAuthData(data), pulsar::ResultOk);
ASSERT_EQ(data->hasDataForTls(), true);
- ASSERT_EQ(data->getTlsCertificates(), tlsCertFile);
- ASSERT_EQ(data->getTlsPrivateKey(), tlsKeyFile);
+ ASSERT_EQ(data->getTlsCertificates(), clientPublicKeyPath);
+ ASSERT_EQ(data->getTlsPrivateKey(), clientPrivateKeyPath);
ClientConfiguration config = ClientConfiguration();
config.setAuth(auth);
-
config.setTlsTrustCertsFilePath("../../pulsar-broker/src/test/resources/authentication/tls/cacert.pem");
+ config.setTlsTrustCertsFilePath(caPath);
config.setTlsAllowInsecureConnection(false);
- Client client("pulsar+ssl://localhost:9886", config);
+ Client client(serviceUrlTls, config);
- std::string topicName =
"persistent://property/cluster/namespace/test-tls-factory";
+ std::string topicName = "persistent://private/auth/test-tls-factory";
Producer producer;
Promise<Result, Producer> producerPromise;
client.createProducerAsync(topicName,
WaitForCallbackValue<Producer>(producerPromise));
@@ -263,15 +269,17 @@ TEST(AuthPluginTest, testAuthFactoryTls) {
}
TEST(AuthPluginTest, testAuthFactoryAthenz) {
- boost::thread zts(&testAthenz::mockZTS);
+ boost::thread zts(boost::bind(&testAthenz::mockZTS, 9998));
pulsar::AuthenticationDataPtr data;
std::string params = R"({
- "tenantDomain": "pulsar.test.tenant",
+ "tenantDomain": "pulsar.test2.tenant",
"tenantService": "service",
"providerDomain": "pulsar.test.provider",
- "privateKey":
"file:../../pulsar-broker/src/test/resources/authentication/tls/client-key.pem",
- "ztsUrl": "http://localhost:9999";
+ "privateKey": "file:)" +
+ clientPrivateKeyPath + R"(",
+ "ztsUrl": "http://localhost:9998";
})";
+ LOG_INFO("PARAMS: " << params);
pulsar::AuthenticationPtr auth = pulsar::AuthFactory::create("athenz",
params);
ASSERT_EQ(auth->getAuthMethodName(), "athenz");
ASSERT_EQ(auth->getAuthData(data), pulsar::ResultOk);
@@ -279,14 +287,18 @@ TEST(AuthPluginTest, testAuthFactoryAthenz) {
ASSERT_EQ(data->hasDataFromCommand(), true);
ASSERT_EQ(data->getHttpHeaders(), "Athenz-Role-Auth: mockToken");
ASSERT_EQ(data->getCommandData(), "mockToken");
+
+ LOG_INFO("Calling zts.join()");
zts.join();
+ LOG_INFO("Done zts.join()");
+
std::vector<std::string> kvs;
boost::algorithm::split(kvs, testAthenz::principalToken,
boost::is_any_of(";"));
for (std::vector<std::string>::iterator itr = kvs.begin(); itr !=
kvs.end(); itr++) {
std::vector<std::string> kv;
boost::algorithm::split(kv, *itr, boost::is_any_of("="));
if (kv[0] == "d") {
- ASSERT_EQ(kv[1], "pulsar.test.tenant");
+ ASSERT_EQ(kv[1], "pulsar.test2.tenant");
} else if (kv[0] == "n") {
ASSERT_EQ(kv[1], "service");
}
diff --git a/pulsar-client-cpp/tests/BasicEndToEndTest.cc
b/pulsar-client-cpp/tests/BasicEndToEndTest.cc
index d264ab129c..b88d8a5ce0 100644
--- a/pulsar-client-cpp/tests/BasicEndToEndTest.cc
+++ b/pulsar-client-cpp/tests/BasicEndToEndTest.cc
@@ -45,8 +45,8 @@ boost::mutex mutex_;
static int globalTestBatchMessagesCounter = 0;
static int globalCount = 0;
static long globalResendMessageCount = 0;
-static std::string lookupUrl = "pulsar://localhost:8885";
-static std::string adminUrl = "http://localhost:8765/";;
+static std::string lookupUrl = "pulsar://localhost:6650";
+static std::string adminUrl = "http://localhost:8080/";;
static void messageListenerFunction(Consumer consumer, const Message& msg) {
globalCount++;
consumer.acknowledge(msg);
@@ -113,7 +113,7 @@ class EncKeyReader : public CryptoKeyReader {
TEST(BasicEndToEndTest, testBatchMessages) {
ClientConfiguration config;
Client client(lookupUrl);
- std::string topicName =
"persistent://property/cluster/namespace/test-batch-messages";
+ std::string topicName = "persistent://public/default/test-batch-messages";
std::string subName = "subscription-name";
Producer producer;
@@ -200,7 +200,7 @@ void resendMessage(Result r, const Message msg, Producer
producer) {
TEST(BasicEndToEndTest, testProduceConsume) {
ClientConfiguration config;
Client client(lookupUrl);
- std::string topicName = "persistent://prop/unit/ns1/my-topic";
+ std::string topicName = "persistent://public/default/test-produce-consume";
std::string subName = "my-sub-name";
Producer producer;
@@ -238,7 +238,7 @@ TEST(BasicEndToEndTest, testProduceConsume) {
}
TEST(BasicEndToEndTest, testLookupThrottling) {
- std::string topicName = "persistent://prop/unit/ns1/testLookupThrottling";
+ std::string topicName = "testLookupThrottling";
ClientConfiguration config;
config.setConcurrentLookupRequest(0);
Client client(lookupUrl, config);
@@ -264,7 +264,7 @@ TEST(BasicEndToEndTest, testNonExistingTopic) {
}
TEST(BasicEndToEndTest, testNonPersistentTopic) {
- std::string topicName =
"non-persistent://prop/unit/ns1/testNonPersistentTopic";
+ std::string topicName =
"non-persistent://public/default/testNonPersistentTopic";
Client client(lookupUrl);
Producer producer;
Result result = client.createProducer(topicName, producer);
@@ -306,7 +306,7 @@ TEST(BasicEndToEndTest, testV2TopicHttp) {
}
TEST(BasicEndToEndTest, testSingleClientMultipleSubscriptions) {
- std::string topicName =
"persistent://prop/unit/ns1/testSingleClientMultipleSubscriptions";
+ std::string topicName = "testSingleClientMultipleSubscriptions";
Client client(lookupUrl);
@@ -325,7 +325,7 @@ TEST(BasicEndToEndTest,
testSingleClientMultipleSubscriptions) {
}
TEST(BasicEndToEndTest, testMultipleClientsMultipleSubscriptions) {
- std::string topicName =
"persistent://prop/unit/ns1/testMultipleClientsMultipleSubscriptions";
+ std::string topicName = "testMultipleClientsMultipleSubscriptions";
Client client1(lookupUrl);
Client client2(lookupUrl);
@@ -354,7 +354,7 @@ TEST(BasicEndToEndTest,
testMultipleClientsMultipleSubscriptions) {
}
TEST(BasicEndToEndTest, testProduceAndConsumeAfterClientClose) {
- std::string topicName =
"persistent://prop/unit/ns1/testProduceAndConsumeAfterClientClose";
+ std::string topicName = "testProduceAndConsumeAfterClientClose";
Client client(lookupUrl);
Producer producer;
@@ -410,18 +410,18 @@ TEST(BasicEndToEndTest,
testProduceAndConsumeAfterClientClose) {
TEST(BasicEndToEndTest, testIamSoFancyCharactersInTopicName) {
Client client(lookupUrl);
Producer producer;
- Result result =
client.createProducer("persistent://prop/unit/ns1/topic@%*)(&!%$#@#$><?",
producer);
+ Result result =
client.createProducer("persistent://public/default/topic@%*)(&!%$#@#$><?",
producer);
ASSERT_EQ(ResultOk, result);
Consumer consumer;
- result =
client.subscribe("persistent://prop/unit/ns1/topic@%*)(&!%$#@#$><?",
"my-sub-name", consumer);
+ result =
client.subscribe("persistent://public/default/topic@%*)(&!%$#@#$><?",
"my-sub-name", consumer);
ASSERT_EQ(ResultOk, result);
}
TEST(BasicEndToEndTest, testSubscribeCloseUnsubscribeSherpaScenario) {
ClientConfiguration config;
Client client(lookupUrl, config);
- std::string topicName = "persistent://prop/unit/ns1/::,::bf11";
+ std::string topicName = "persistent://public/default/::,::bf11";
std::string subName = "weird-ass-characters-@%*)(&!%$#@#$><?)";
Producer producer;
Result result = client.createProducer(topicName, producer);
@@ -442,7 +442,7 @@ TEST(BasicEndToEndTest,
testSubscribeCloseUnsubscribeSherpaScenario) {
TEST(BasicEndToEndTest, testInvalidUrlPassed) {
Client client("localhost:4080");
- std::string topicName = "persistent://prop/unit/ns1/testInvalidUrlPassed";
+ std::string topicName = "testInvalidUrlPassed";
std::string subName = "test-sub";
Producer producer;
Result result = client.createProducer(topicName, producer);
@@ -467,10 +467,11 @@ TEST(BasicEndToEndTest, testInvalidUrlPassed) {
TEST(BasicEndToEndTest, testPartitionedProducerConsumer) {
Client client(lookupUrl);
- std::string topicName =
"persistent://prop/unit/ns/testPartitionedProducerConsumer";
+ std::string topicName = "testPartitionedProducerConsumer";
// call admin api to make it partitioned
- std::string url = adminUrl +
"admin/persistent/prop/unit/ns/testPartitionedProducerConsumer/partitions";
+ std::string url =
+ adminUrl +
"admin/v2/persistent/public/default/testPartitionedProducerConsumer/partitions";
int res = makePutRequest(url, "3");
LOG_INFO("res = " << res);
@@ -506,11 +507,12 @@ TEST(BasicEndToEndTest, testPartitionedProducerConsumer) {
TEST(BasicEndToEndTest, testPartitionedProducerConsumerSubscriptionName) {
Client client(lookupUrl);
- std::string topicName =
"persistent://prop/unit/ns/testPartitionedProducerConsumerSubscriptionName";
+ std::string topicName = "testPartitionedProducerConsumerSubscriptionName";
// call admin api to make it partitioned
std::string url =
- adminUrl +
"admin/persistent/prop/unit/ns/testPartitionedProducerConsumerSubscriptionName/partitions";
+ adminUrl +
+
"admin/v2/persistent/public/default/testPartitionedProducerConsumerSubscriptionName/partitions";
int res = makePutRequest(url, "3");
LOG_INFO("res = " << res);
@@ -531,7 +533,7 @@ TEST(BasicEndToEndTest,
testPartitionedProducerConsumerSubscriptionName) {
TEST(BasicEndToEndTest, testMessageTooBig) {
ClientConfiguration config;
Client client(lookupUrl);
- std::string topicName = "persistent://prop/unit/ns1/testMessageTooBig";
+ std::string topicName = "testMessageTooBig";
Producer producer;
Result result = client.createProducer(topicName, producer);
ASSERT_EQ(ResultOk, result);
@@ -554,7 +556,7 @@ TEST(BasicEndToEndTest, testMessageTooBig) {
TEST(BasicEndToEndTest, testCompressionLZ4) {
ClientConfiguration config;
Client client(lookupUrl);
- std::string topicName =
"persistent://prop/unit/namespace1/testCompressionLZ4";
+ std::string topicName = "testCompressionLZ4";
std::string subName = "my-sub-name";
Producer producer;
ProducerConfiguration conf;
@@ -592,7 +594,7 @@ TEST(BasicEndToEndTest, testCompressionLZ4) {
TEST(BasicEndToEndTest, testCompressionZLib) {
ClientConfiguration config;
Client client(lookupUrl);
- std::string topicName = "persistent://prop/unit/ns1/testCompressionZLib";
+ std::string topicName = "testCompressionZLib";
std::string subName = "my-sub-name";
Producer producer;
ProducerConfiguration conf;
@@ -645,11 +647,11 @@ TEST(BasicEndToEndTest, testConfigurationFile) {
TEST(BasicEndToEndTest, testSinglePartitionRoutingPolicy) {
Client client(lookupUrl);
- std::string topicName =
"persistent://prop/unit/ns/partition-testSinglePartitionRoutingPolicy";
+ std::string topicName = "partition-testSinglePartitionRoutingPolicy";
// call admin api to make it partitioned
std::string url =
- adminUrl +
"admin/persistent/prop/unit/ns/partition-testSinglePartitionRoutingPolicy/partitions";
+ adminUrl +
"admin/v2/persistent/public/default/partition-testSinglePartitionRoutingPolicy/partitions";
int res = makePutRequest(url, "5");
LOG_INFO("res = " << res);
@@ -695,7 +697,7 @@ TEST(BasicEndToEndTest, testNamespaceName) {
TEST(BasicEndToEndTest, testConsumerClose) {
ClientConfiguration config;
Client client(lookupUrl);
- std::string topicName = "persistent://prop/unit/ns1/testConsumerClose";
+ std::string topicName = "testConsumerClose";
std::string subName = "my-sub-name";
Consumer consumer;
ASSERT_EQ(ResultOk, client.subscribe(topicName, subName, consumer));
@@ -705,12 +707,12 @@ TEST(BasicEndToEndTest, testConsumerClose) {
TEST(BasicEndToEndTest, testDuplicateConsumerCreationOnPartitionedTopic) {
Client client(lookupUrl);
- std::string topicName =
-
"persistent://prop/unit/ns/partition-testDuplicateConsumerCreationOnPartitionedTopic";
+ std::string topicName =
"partition-testDuplicateConsumerCreationOnPartitionedTopic";
// call admin api to make it partitioned
std::string url =
- adminUrl +
"admin/persistent/prop/unit/ns/testDuplicateConsumerCreationOnPartitionedTopic/partitions";
+ adminUrl +
+
"admin/v2/persistent/public/default/testDuplicateConsumerCreationOnPartitionedTopic/partitions";
int res = makePutRequest(url, "5");
LOG_INFO("res = " << res);
@@ -755,10 +757,10 @@ TEST(BasicEndToEndTest,
testDuplicateConsumerCreationOnPartitionedTopic) {
TEST(BasicEndToEndTest, testRoundRobinRoutingPolicy) {
Client client(lookupUrl);
- std::string topicName =
"persistent://prop/unit/ns/partition-testRoundRobinRoutingPolicy";
+ std::string topicName =
"persistent://public/default/partition-testRoundRobinRoutingPolicy";
// call admin api to make it partitioned
std::string url =
- adminUrl +
"admin/persistent/prop/unit/ns/partition-testRoundRobinRoutingPolicy/partitions";
+ adminUrl +
"admin/v2/persistent/public/default/partition-testRoundRobinRoutingPolicy/partitions";
int res = makePutRequest(url, "5");
LOG_INFO("res = " << res);
@@ -820,9 +822,10 @@ TEST(BasicEndToEndTest, testRoundRobinRoutingPolicy) {
TEST(BasicEndToEndTest, testMessageListener) {
Client client(lookupUrl);
- std::string topicName =
"persistent://prop/unit/ns/partition-testMessageListener";
+ std::string topicName = "partition-testMessageListener";
// call admin api to make it partitioned
- std::string url = adminUrl +
"admin/persistent/prop/unit/ns/partition-testMessageListener/partitions";
+ std::string url =
+ adminUrl +
"admin/v2/persistent/public/default/partition-testMessageListener/partitions";
int res = makePutRequest(url, "5");
LOG_INFO("res = " << res);
@@ -861,12 +864,11 @@ TEST(BasicEndToEndTest, testMessageListener) {
TEST(BasicEndToEndTest, testMessageListenerPause) {
Client client(lookupUrl);
- std::string topicName =
"persistent://property/cluster/namespace/partition-testMessageListenerPause";
+ std::string topicName = "partition-testMessageListenerPause";
// call admin api to make it partitioned
std::string url =
- adminUrl +
-
"admin/persistent/property/cluster/namespace/partition-testMessageListener-pauses/partitions";
+ adminUrl +
"admin/v2/persistent/public/default/partition-testMessageListener-pauses/partitions";
int res = makePutRequest(url, "5");
LOG_INFO("res = " << res);
@@ -919,7 +921,7 @@ TEST(BasicEndToEndTest, testResendViaSendCallback) {
ClientConfiguration clientConfiguration;
clientConfiguration.setIOThreads(1);
Client client(lookupUrl, clientConfiguration);
- std::string topicName =
"persistent://my-property/my-cluster/my-namespace/testResendViaListener";
+ std::string topicName = "testResendViaListener";
Producer producer;
@@ -953,7 +955,7 @@ TEST(BasicEndToEndTest, testStatsLatencies) {
config.setMessageListenerThreads(1);
config.setStatsIntervalInSeconds(5);
Client client(lookupUrl, config);
- std::string topicName =
"persistent://property/cluster/namespace/testStatsLatencies";
+ std::string topicName = "persistent://public/default/testStatsLatencies";
std::string subName = "subscription-name";
Producer producer;
@@ -1066,7 +1068,7 @@ TEST(BasicEndToEndTest, testStatsLatencies) {
TEST(BasicEndToEndTest, testProduceMessageSize) {
ClientConfiguration config;
Client client(lookupUrl);
- std::string topicName =
"persistent://prop/unit/ns1/testProduceMessageSize";
+ std::string topicName = "testProduceMessageSize";
std::string subName = "my-sub-name";
Producer producer1;
Producer producer2;
@@ -1117,7 +1119,7 @@ TEST(BasicEndToEndTest, testProduceMessageSize) {
TEST(BasicEndToEndTest, testHandlerReconnectionLogic) {
Client client(adminUrl);
- std::string topicName =
"persistent://prop/unit/ns1/testHandlerReconnectionLogic";
+ std::string topicName = "testHandlerReconnectionLogic";
Producer producer;
Consumer consumer;
@@ -1183,7 +1185,7 @@ TEST(BasicEndToEndTest, testHandlerReconnectionLogic) {
TEST(BasicEndToEndTest, testRSAEncryption) {
ClientConfiguration config;
Client client(lookupUrl);
- std::string topicName = "persistent://prop/unit/ns1/my-rsaenctopic";
+ std::string topicName = "my-rsaenctopic";
std::string subName = "my-sub-name";
Producer producer;
@@ -1241,7 +1243,7 @@ TEST(BasicEndToEndTest, testRSAEncryption) {
TEST(BasicEndToEndTest, testEncryptionFailure) {
ClientConfiguration config;
Client client(lookupUrl);
- std::string topicName = "persistent://prop/unit/ns1/my-rsaencfailtopic";
+ std::string topicName = "my-rsaencfailtopic";
std::string subName = "my-sub-name";
Producer producer;
@@ -1355,7 +1357,7 @@ TEST(BasicEndToEndTest, testEncryptionFailure) {
TEST(BasicEndToEndTest, testEventTime) {
ClientConfiguration config;
Client client(lookupUrl, config);
- std::string topicName = "persistent://prop/unit/ns1/topic";
+ std::string topicName = "test-event-time";
Producer producer;
ProducerConfiguration producerConf;
producerConf.setBatchingEnabled(true);
@@ -1381,7 +1383,7 @@ TEST(BasicEndToEndTest, testEventTime) {
TEST(BasicEndToEndTest, testSeek) {
ClientConfiguration config;
Client client(lookupUrl);
- std::string topicName = "persistent://prop/unit/ns1/testSeek";
+ std::string topicName = "persistent://public/default/testSeek";
std::string subName = "sub-testSeek";
Producer producer;
@@ -1448,7 +1450,7 @@ TEST(BasicEndToEndTest, testSeek) {
TEST(BasicEndToEndTest, testUnAckedMessageTimeout) {
Client client(lookupUrl);
- std::string topicName =
"persistent://prop/unit/ns1/testUnAckedMessageTimeout";
+ std::string topicName = "testUnAckedMessageTimeout";
std::string subName = "my-sub-name";
std::string content = "msg-content";
@@ -1488,7 +1490,7 @@ TEST(BasicEndToEndTest, testUnAckedMessageTimeout) {
TEST(BasicEndToEndTest, testUnAckedMessageTimeoutListener) {
Client client(lookupUrl);
- std::string topicName =
"persistent://prop/unit/ns1/testUnAckedMessageTimeoutListener";
+ std::string topicName = "testUnAckedMessageTimeoutListener";
std::string subName = "my-sub-name";
std::string content = "msg-content";
@@ -1525,7 +1527,7 @@ TEST(BasicEndToEndTest,
testMultiTopicsConsumerTopicNameInvalid) {
topicNames.reserve(3);
std::string subName = "testMultiTopicsTopicNameInvalid";
// cluster empty
- std::string topicName1 =
"persistent://prop/testMultiTopicsTopicNameInvalid";
+ std::string topicName1 =
"persistent://tenant/testMultiTopicsTopicNameInvalid";
// empty topics
ASSERT_EQ(0, topicNames.size());
@@ -1560,9 +1562,9 @@ TEST(BasicEndToEndTest,
testMultiTopicsConsumerDifferentNamespace) {
std::vector<std::string> topicNames;
topicNames.reserve(3);
std::string subName = "testMultiTopicsDifferentNamespace";
- std::string topicName1 =
"persistent://prop/unit/ns1/testMultiTopicsConsumerDifferentNamespace1";
- std::string topicName2 =
"persistent://prop/unit/ns2/testMultiTopicsConsumerDifferentNamespace2";
- std::string topicName3 =
"persistent://prop/unit/ns3/testMultiTopicsConsumerDifferentNamespace3";
+ std::string topicName1 =
"persistent://public/default/testMultiTopicsConsumerDifferentNamespace1";
+ std::string topicName2 =
"persistent://public/default-2/testMultiTopicsConsumerDifferentNamespace2";
+ std::string topicName3 =
"persistent://public/default-3/testMultiTopicsConsumerDifferentNamespace3";
topicNames.push_back(topicName1);
topicNames.push_back(topicName2);
@@ -1570,11 +1572,13 @@ TEST(BasicEndToEndTest,
testMultiTopicsConsumerDifferentNamespace) {
// call admin api to make topics partitioned
std::string url1 =
- adminUrl +
"admin/persistent/prop/unit/ns1/testMultiTopicsConsumerDifferentNamespace1/partitions";
+ adminUrl +
"admin/v2/persistent/public/default/testMultiTopicsConsumerDifferentNamespace1/partitions";
std::string url2 =
- adminUrl +
"admin/persistent/prop/unit/ns2/testMultiTopicsConsumerDifferentNamespace2/partitions";
+ adminUrl +
+
"admin/v2/persistent/public/default-2/testMultiTopicsConsumerDifferentNamespace2/partitions";
std::string url3 =
- adminUrl +
"admin/persistent/prop/unit/ns3/testMultiTopicsConsumerDifferentNamespace3/partitions";
+ adminUrl +
+
"admin/v2/persistent/public/default-3/testMultiTopicsConsumerDifferentNamespace3/partitions";
int res = makePutRequest(url1, "2");
ASSERT_FALSE(res != 204 && res != 409);
@@ -1604,10 +1608,10 @@ TEST(BasicEndToEndTest, testMultiTopicsConsumerPubSub) {
std::vector<std::string> topicNames;
topicNames.reserve(3);
std::string subName = "testMultiTopicsConsumer";
- std::string topicName1 =
"persistent://prop/unit/ns/testMultiTopicsConsumer1";
- std::string topicName2 =
"persistent://prop/unit/ns/testMultiTopicsConsumer2";
- std::string topicName3 =
"persistent://prop/unit/ns/testMultiTopicsConsumer3";
- std::string topicName4 =
"persistent://prop/unit/ns/testMultiTopicsConsumer4";
+ std::string topicName1 = "testMultiTopicsConsumer1";
+ std::string topicName2 = "testMultiTopicsConsumer2";
+ std::string topicName3 = "testMultiTopicsConsumer3";
+ std::string topicName4 = "testMultiTopicsConsumer4";
topicNames.push_back(topicName1);
topicNames.push_back(topicName2);
@@ -1615,9 +1619,9 @@ TEST(BasicEndToEndTest, testMultiTopicsConsumerPubSub) {
topicNames.push_back(topicName4);
// call admin api to make topics partitioned
- std::string url1 = adminUrl +
"admin/persistent/prop/unit/ns/testMultiTopicsConsumer1/partitions";
- std::string url2 = adminUrl +
"admin/persistent/prop/unit/ns/testMultiTopicsConsumer2/partitions";
- std::string url3 = adminUrl +
"admin/persistent/prop/unit/ns/testMultiTopicsConsumer3/partitions";
+ std::string url1 = adminUrl +
"admin/v2/persistent/public/default/testMultiTopicsConsumer1/partitions";
+ std::string url2 = adminUrl +
"admin/v2/persistent/public/default/testMultiTopicsConsumer2/partitions";
+ std::string url3 = adminUrl +
"admin/v2/persistent/public/default/testMultiTopicsConsumer3/partitions";
int res = makePutRequest(url1, "2");
ASSERT_FALSE(res != 204 && res != 409);
@@ -1727,24 +1731,24 @@ TEST(BasicEndToEndTest,
testPatternTopicsConsumerInvalid) {
// and only receive messages from matched topics.
TEST(BasicEndToEndTest, testPatternMultiTopicsConsumerPubSub) {
Client client(lookupUrl);
- std::string pattern =
"persistent://prop/unit/ns1/patternMultiTopicsConsumer.*";
+ std::string pattern =
"persistent://public/default/patternMultiTopicsConsumer.*";
std::string subName = "testPatternMultiTopicsConsumer";
- std::string topicName1 =
"persistent://prop/unit/ns1/patternMultiTopicsConsumerPubSub1";
- std::string topicName2 =
"persistent://prop/unit/ns1/patternMultiTopicsConsumerPubSub2";
- std::string topicName3 =
"persistent://prop/unit/ns1/patternMultiTopicsConsumerPubSub3";
+ std::string topicName1 =
"persistent://public/default/patternMultiTopicsConsumerPubSub1";
+ std::string topicName2 =
"persistent://public/default/patternMultiTopicsConsumerPubSub2";
+ std::string topicName3 =
"persistent://public/default/patternMultiTopicsConsumerPubSub3";
// This will not match pattern
- std::string topicName4 =
"persistent://prop/unit/ns1/patternMultiTopicsNotMatchPubSub4";
+ std::string topicName4 =
"persistent://public/default/patternMultiTopicsNotMatchPubSub4";
// call admin api to make topics partitioned
std::string url1 =
- adminUrl +
"admin/persistent/prop/unit/ns1/patternMultiTopicsConsumerPubSub1/partitions";
+ adminUrl +
"admin/v2/persistent/public/default/patternMultiTopicsConsumerPubSub1/partitions";
std::string url2 =
- adminUrl +
"admin/persistent/prop/unit/ns1/patternMultiTopicsConsumerPubSub2/partitions";
+ adminUrl +
"admin/v2/persistent/public/default/patternMultiTopicsConsumerPubSub2/partitions";
std::string url3 =
- adminUrl +
"admin/persistent/prop/unit/ns1/patternMultiTopicsConsumerPubSub3/partitions";
+ adminUrl +
"admin/v2/persistent/public/default/patternMultiTopicsConsumerPubSub3/partitions";
std::string url4 =
- adminUrl +
"admin/persistent/prop/unit/ns1/patternMultiTopicsNotMatchPubSub4/partitions";
+ adminUrl +
"admin/v2/persistent/public/default/patternMultiTopicsNotMatchPubSub4/partitions";
int res = makePutRequest(url1, "2");
ASSERT_FALSE(res != 204 && res != 409);
@@ -1843,7 +1847,7 @@ TEST(BasicEndToEndTest,
testPatternMultiTopicsConsumerPubSub) {
// and only receive messages from matched topics.
TEST(BasicEndToEndTest, testPatternMultiTopicsConsumerAutoDiscovery) {
Client client(lookupUrl);
- std::string pattern =
"persistent://prop/unit/ns2/patternTopicsAutoConsumer.*";
+ std::string pattern =
"persistent://public/default/patternTopicsAutoConsumer.*";
Result result;
std::string subName = "testPatternTopicsAutoConsumer";
@@ -1863,21 +1867,21 @@ TEST(BasicEndToEndTest,
testPatternMultiTopicsConsumerAutoDiscovery) {
LOG_INFO("created pattern consumer with not match topics at beginning");
// 2. create 4 topics, in which 3 match the pattern.
- std::string topicName1 =
"persistent://prop/unit/ns2/patternTopicsAutoConsumerPubSub1";
- std::string topicName2 =
"persistent://prop/unit/ns2/patternTopicsAutoConsumerPubSub2";
- std::string topicName3 =
"persistent://prop/unit/ns2/patternTopicsAutoConsumerPubSub3";
+ std::string topicName1 =
"persistent://public/default/patternTopicsAutoConsumerPubSub1";
+ std::string topicName2 =
"persistent://public/default/patternTopicsAutoConsumerPubSub2";
+ std::string topicName3 =
"persistent://public/default/patternTopicsAutoConsumerPubSub3";
// This will not match pattern
- std::string topicName4 =
"persistent://prop/unit/ns2/patternMultiTopicsNotMatchPubSub4";
+ std::string topicName4 =
"persistent://public/default/patternMultiTopicsNotMatchPubSub4";
// call admin api to make topics partitioned
std::string url1 =
- adminUrl +
"admin/persistent/prop/unit/ns2/patternTopicsAutoConsumerPubSub1/partitions";
+ adminUrl +
"admin/v2/persistent/public/default/patternTopicsAutoConsumerPubSub1/partitions";
std::string url2 =
- adminUrl +
"admin/persistent/prop/unit/ns2/patternTopicsAutoConsumerPubSub2/partitions";
+ adminUrl +
"admin/v2/persistent/public/default/patternTopicsAutoConsumerPubSub2/partitions";
std::string url3 =
- adminUrl +
"admin/persistent/prop/unit/ns2/patternTopicsAutoConsumerPubSub3/partitions";
+ adminUrl +
"admin/v2/persistent/public/default/patternTopicsAutoConsumerPubSub3/partitions";
std::string url4 =
- adminUrl +
"admin/persistent/prop/unit/ns2/patternMultiTopicsNotMatchPubSub4/partitions";
+ adminUrl +
"admin/v2/persistent/public/default/patternMultiTopicsNotMatchPubSub4/partitions";
int res = makePutRequest(url1, "2");
ASSERT_FALSE(res != 204 && res != 409);
@@ -1964,7 +1968,7 @@ TEST(BasicEndToEndTest,
testPatternMultiTopicsConsumerAutoDiscovery) {
TEST(BasicEndToEndTest, testSyncFlushBatchMessages) {
ClientConfiguration config;
Client client(lookupUrl);
- std::string topicName =
"persistent://property/cluster/namespace/test-flush-batch-messages";
+ std::string topicName = "test-flush-batch-messages-" +
boost::lexical_cast<std::string>(time(NULL));
std::string subName = "subscription-name";
Producer producer;
@@ -2013,7 +2017,7 @@ TEST(BasicEndToEndTest, testSyncFlushBatchMessages) {
// message not reached max batch number, should not receive any data.
Message receivedMsg;
- ASSERT_EQ(ResultTimeout, consumer.receive(receivedMsg, 5000));
+ ASSERT_EQ(ResultTimeout, consumer.receive(receivedMsg, 1000));
// Send Asynchronously of the other half the messages
for (int i = numOfMessages / 2; i < numOfMessages; i++) {
@@ -2027,19 +2031,21 @@ TEST(BasicEndToEndTest, testSyncFlushBatchMessages) {
}
LOG_INFO("sending the other half messages in async, should able to
receive");
// message not reached max batch number, should received the messages
- ASSERT_EQ(ResultOk, consumer.receive(receivedMsg, 5000));
+ ASSERT_EQ(ResultOk, consumer.receive(receivedMsg, 1000));
+ LOG_INFO("Receive all messages");
// receive all the messages.
int i = 1;
- while (consumer.receive(receivedMsg, 5000) == ResultOk) {
+ while (consumer.receive(receivedMsg, 1000) == ResultOk) {
std::string expectedMessageContent = prefix +
boost::lexical_cast<std::string>(i);
- LOG_DEBUG("Received Message with [ content - " <<
receivedMsg.getDataAsString() << "] [ messageID = "
- <<
receivedMsg.getMessageId() << "]");
+ LOG_INFO("Received Message with [ content - "
+ << receivedMsg.getDataAsString() << "] [ messageID = " <<
receivedMsg.getMessageId() << "]");
ASSERT_EQ(receivedMsg.getProperty("msgIndex"),
boost::lexical_cast<std::string>(i++));
ASSERT_EQ(expectedMessageContent, receivedMsg.getDataAsString());
ASSERT_EQ(ResultOk, consumer.acknowledge(receivedMsg));
}
+ LOG_INFO("Last sync send round");
// Send sync of half the messages, this will triggerFlush, and could get
the messages.
prefix = "msg-batch-sync";
for (int i = 0; i < numOfMessages / 2; i++) {
@@ -2049,10 +2055,10 @@ TEST(BasicEndToEndTest, testSyncFlushBatchMessages) {
.setProperty("msgIndex",
boost::lexical_cast<std::string>(i))
.build();
producer.send(msg);
- LOG_DEBUG("sync sending message " << messageContent);
+ LOG_INFO("sync sending message " << messageContent);
}
// message not reached max batch number, should received the messages, and
not timeout
- ASSERT_EQ(ResultOk, consumer.receive(receivedMsg, 5000));
+ ASSERT_EQ(ResultOk, consumer.receive(receivedMsg, 1000));
producer.close();
client.shutdown();
@@ -2065,10 +2071,10 @@ static void simpleCallback(Result code, const Message&
msg) {
TEST(BasicEndToEndTest, testSyncFlushBatchMessagesPartitionedTopic) {
Client client(lookupUrl);
- std::string topicName =
"persistent://prop/unit/ns/partition-testSyncFlushBatchMessages";
+ std::string topicName =
"persistent://public/default/partition-testSyncFlushBatchMessages";
// call admin api to make it partitioned
std::string url =
- adminUrl +
"admin/persistent/prop/unit/ns/partition-testSyncFlushBatchMessages/partitions";
+ adminUrl +
"admin/v2/persistent/public/default/partition-testSyncFlushBatchMessages/partitions";
int res = makePutRequest(url, "5");
int numberOfPartitions = 5;
diff --git a/pulsar-client-cpp/tests/BatchMessageTest.cc
b/pulsar-client-cpp/tests/BatchMessageTest.cc
index 894a582903..d142b63684 100644
--- a/pulsar-client-cpp/tests/BatchMessageTest.cc
+++ b/pulsar-client-cpp/tests/BatchMessageTest.cc
@@ -41,8 +41,8 @@ using namespace pulsar;
static int globalTestBatchMessagesCounter = 0;
static int globalCount = 0;
-static std::string lookupUrl = "pulsar://localhost:8885";
-static std::string adminUrl = "http://localhost:8765/";;
+static std::string lookupUrl = "pulsar://localhost:6650";
+static std::string adminUrl = "http://localhost:8080/";;
// ecpoch time in seconds
long epochTime = time(NULL);
@@ -87,7 +87,7 @@ TEST(BatchMessageTest, testProducerTimeout) {
clientConf.setStatsIntervalInSeconds(1);
Client client(lookupUrl, clientConf);
- std::string topicName = "persistent://property/cluster/namespace/" +
testName;
+ std::string topicName = "persistent://public/default/" + testName;
std::string subName = "subscription-name";
Producer producer;
@@ -172,7 +172,7 @@ TEST(BatchMessageTest, testBatchSizeInBytes) {
globalTestBatchMessagesCounter = 0;
Client client(lookupUrl);
- std::string topicName = "persistent://property/cluster/namespace/" +
testName;
+ std::string topicName = "persistent://public/default/" + testName;
std::string subName = "subscription-name";
Producer producer;
@@ -256,7 +256,7 @@ TEST(BatchMessageTest, testSmallReceiverQueueSize) {
clientConf.setStatsIntervalInSeconds(20);
Client client(lookupUrl, clientConf);
- std::string topicName = "persistent://property/cluster/namespace/" +
testName;
+ std::string topicName = "persistent://public/default/" + testName;
std::string subName = "subscription-name";
Producer producer;
@@ -354,7 +354,7 @@ TEST(BatchMessageTest, testIndividualAck) {
clientConfig.setStatsIntervalInSeconds(1);
Client client(lookupUrl, clientConfig);
- std::string topicName = "persistent://property/cluster/namespace/" +
testName;
+ std::string topicName = "persistent://public/default/" + testName;
std::string subName = "subscription-name";
Producer producer;
@@ -509,7 +509,7 @@ TEST(BatchMessageTest, testCumulativeAck) {
clientConfig.setStatsIntervalInSeconds(100);
Client client(lookupUrl, clientConfig);
- std::string topicName = "persistent://property/cluster/namespace/" +
testName;
+ std::string topicName = "persistent://public/default/" + testName;
std::string subName = "subscription-name";
Producer producer;
@@ -635,7 +635,7 @@ TEST(BatchMessageTest, testMixedAck) {
std::string testName = boost::lexical_cast<std::string>(epochTime) +
"testMixedAck";
Client client(lookupUrl);
- std::string topicName = "persistent://property/cluster/namespace/" +
testName;
+ std::string topicName = "persistent://public/default/" + testName;
std::string subName = "subscription-name";
Producer producer;
@@ -743,7 +743,7 @@ TEST(BatchMessageTest, testPermits) {
std::string testName = boost::lexical_cast<std::string>(epochTime) +
"testPermits";
Client client(lookupUrl);
- std::string topicName = "persistent://property/cluster/namespace/" +
testName;
+ std::string topicName = "persistent://public/default/" + testName;
std::string subName = "subscription-name";
Producer producer;
@@ -855,12 +855,11 @@ TEST(BatchMessageTest, testPermits) {
TEST(BatchMessageTest, testPartitionedTopics) {
Client client(lookupUrl);
- std::string topicName =
"persistent://property/cluster/namespace/test-partitioned-batch-messages-" +
+ std::string topicName =
"persistent://public/default/test-partitioned-batch-messages-" +
boost::lexical_cast<std::string>(epochTime);
// call admin api to make it partitioned
- std::string url = adminUrl +
-
"admin/persistent/property/cluster/namespace/test-partitioned-batch-messages-" +
+ std::string url = adminUrl +
"admin/v2/persistent/public/default/test-partitioned-batch-messages-" +
boost::lexical_cast<std::string>(epochTime) +
"/partitions";
int res = makePutRequest(url, "7");
diff --git a/pulsar-client-cpp/tests/BinaryLookupServiceTest.cc
b/pulsar-client-cpp/tests/BinaryLookupServiceTest.cc
index f706eb8575..8fd0354a11 100644
--- a/pulsar-client-cpp/tests/BinaryLookupServiceTest.cc
+++ b/pulsar-client-cpp/tests/BinaryLookupServiceTest.cc
@@ -36,14 +36,13 @@ using namespace pulsar;
TEST(BinaryLookupServiceTest, basicLookup) {
ExecutorServiceProviderPtr service =
boost::make_shared<ExecutorServiceProvider>(1);
AuthenticationPtr authData = AuthFactory::Disabled();
- std::string url = "pulsar://localhost:8885";
+ std::string url = "pulsar://localhost:6650";
ClientConfiguration conf;
ExecutorServiceProviderPtr
ioExecutorProvider_(boost::make_shared<ExecutorServiceProvider>(1));
ConnectionPool pool_(conf, ioExecutorProvider_, authData, true);
BinaryProtoLookupService lookupService(pool_, url);
- std::string topic = "persistent://prop/unit/ns1/topic";
- TopicNamePtr topicName = TopicName::get(topic);
+ TopicNamePtr topicName = TopicName::get("topic");
Future<Result, LookupDataResultPtr> partitionFuture =
lookupService.getPartitionMetadataAsync(topicName);
LookupDataResultPtr lookupData;
@@ -51,8 +50,7 @@ TEST(BinaryLookupServiceTest, basicLookup) {
ASSERT_TRUE(lookupData != NULL);
ASSERT_EQ(0, lookupData->getPartitions());
- Future<Result, LookupDataResultPtr> future =
- lookupService.lookupAsync("persistent://prop/unit/ns1/topic");
+ Future<Result, LookupDataResultPtr> future =
lookupService.lookupAsync("topic");
result = future.get(lookupData);
ASSERT_EQ(ResultOk, result);
@@ -61,22 +59,22 @@ TEST(BinaryLookupServiceTest, basicLookup) {
}
TEST(BinaryLookupServiceTest, basicGetNamespaceTopics) {
- std::string url = "pulsar://localhost:8885";
- std::string adminUrl = "http://localhost:8765/";;
+ std::string url = "pulsar://localhost:6650";
+ std::string adminUrl = "http://localhost:8080/";;
Result result;
// 1. create some topics under same namespace
Client client(url);
- std::string topicName1 =
"persistent://prop/unit/ns4/basicGetNamespaceTopics1";
- std::string topicName2 =
"persistent://prop/unit/ns4/basicGetNamespaceTopics2";
- std::string topicName3 =
"persistent://prop/unit/ns4/basicGetNamespaceTopics3";
+ std::string topicName1 =
"persistent://public/default/basicGetNamespaceTopics1";
+ std::string topicName2 =
"persistent://public/default/basicGetNamespaceTopics2";
+ std::string topicName3 =
"persistent://public/default/basicGetNamespaceTopics3";
// This is not in same namespace.
- std::string topicName4 =
"persistent://prop/unit/ns2/basicGetNamespaceTopics4";
+ std::string topicName4 =
"persistent://public/default-2/basicGetNamespaceTopics4";
// call admin api to make topics partitioned
- std::string url1 = adminUrl +
"admin/persistent/prop/unit/ns4/basicGetNamespaceTopics1/partitions";
- std::string url2 = adminUrl +
"admin/persistent/prop/unit/ns4/basicGetNamespaceTopics2/partitions";
- std::string url3 = adminUrl +
"admin/persistent/prop/unit/ns4/basicGetNamespaceTopics3/partitions";
+ std::string url1 = adminUrl +
"admin/v2/persistent/public/default/basicGetNamespaceTopics1/partitions";
+ std::string url2 = adminUrl +
"admin/v2/persistent/public/default/basicGetNamespaceTopics2/partitions";
+ std::string url3 = adminUrl +
"admin/v2/persistent/public/default/basicGetNamespaceTopics3/partitions";
int res = makePutRequest(url1, "2");
ASSERT_FALSE(res != 204 && res != 409);
@@ -116,10 +114,10 @@ TEST(BinaryLookupServiceTest, basicGetNamespaceTopics) {
ASSERT_TRUE(topicsData != NULL);
// 3. verify result contains first 3 topic
- ASSERT_EQ(topicsData->size(), 3);
ASSERT_TRUE(std::find(topicsData->begin(), topicsData->end(), topicName1)
!= topicsData->end());
ASSERT_TRUE(std::find(topicsData->begin(), topicsData->end(), topicName2)
!= topicsData->end());
ASSERT_TRUE(std::find(topicsData->begin(), topicsData->end(), topicName3)
!= topicsData->end());
+ ASSERT_FALSE(std::find(topicsData->begin(), topicsData->end(), topicName4)
!= topicsData->end());
client.shutdown();
}
diff --git a/pulsar-client-cpp/tests/ClientDeduplicationTest.cc
b/pulsar-client-cpp/tests/ClientDeduplicationTest.cc
index b235040b96..c7beeb042a 100644
--- a/pulsar-client-cpp/tests/ClientDeduplicationTest.cc
+++ b/pulsar-client-cpp/tests/ClientDeduplicationTest.cc
@@ -27,24 +27,31 @@
using namespace pulsar;
-static std::string serviceUrl = "pulsar://localhost:8885";
-static std::string adminUrl = "http://localhost:8765/";;
+static std::string serviceUrl = "pulsar://localhost:6650";
+static std::string adminUrl = "http://localhost:8080/";;
TEST(ClientDeduplicationTest, testProducerSequenceAfterReconnect) {
Client client(serviceUrl);
- std::string topicName =
"persistent://sample/standalone/ns-dedup-1/testProducerSequenceAfterReconnect-"
+
+ std::string topicName =
"persistent://public/dedup-1/testProducerSequenceAfterReconnect-" +
boost::lexical_cast<std::string>(time(NULL));
// call admin api to create namespace and enable deduplication
- std::string url = adminUrl +
"admin/namespaces/sample/standalone/ns-dedup-1";
- int res = makePutRequest(url, "");
+ std::string url = adminUrl + "admin/v2/namespaces/public/dedup-1";
+ int res = makePutRequest(url, R"({"replication_clusters":
["standalone"]})");
ASSERT_TRUE(res == 204 || res == 409);
- url = adminUrl +
"admin/namespaces/sample/standalone/ns-dedup-1/deduplication";
+ url = adminUrl +
"admin/v2/namespaces/public/dedup-1/permissions/anonymous";
+ res = makePostRequest(url, R"(["produce","consume"])");
+ ASSERT_TRUE(res == 204 || res == 409);
+
+ url = adminUrl + "admin/v2/namespaces/public/dedup-1/deduplication";
res = makePostRequest(url, "true");
ASSERT_TRUE(res == 204 || res == 409);
+ // Ensure dedup status was refreshed
+ sleep(1);
+
ReaderConfiguration readerConf;
Reader reader;
ASSERT_EQ(client.createReader(topicName, MessageId::earliest(),
readerConf, reader), ResultOk);
@@ -79,17 +86,20 @@ TEST(ClientDeduplicationTest,
testProducerSequenceAfterReconnect) {
}
TEST(ClientDeduplicationTest, testProducerDeduplication) {
- Client client(serviceUrl);
+ Client client(adminUrl);
- std::string topicName =
"persistent://sample/standalone/ns-dedup-2/testProducerDeduplication-" +
+ std::string topicName =
"persistent://public/dedup-2/testProducerDeduplication-" +
boost::lexical_cast<std::string>(time(NULL));
- // call admin api to create namespace and enable deduplication
- std::string url = adminUrl +
"admin/namespaces/sample/standalone/ns-dedup-2";
- int res = makePutRequest(url, "");
+ std::string url = adminUrl + "admin/v2/namespaces/public/dedup-2";
+ int res = makePutRequest(url, R"({"replication_clusters":
["standalone"]})");
+ ASSERT_TRUE(res == 204 || res == 409);
+
+ url = adminUrl +
"admin/v2/namespaces/public/dedup-2/permissions/anonymous";
+ res = makePostRequest(url, R"(["produce","consume"])");
ASSERT_TRUE(res == 204 || res == 409);
- url = adminUrl +
"admin/namespaces/sample/standalone/ns-dedup-2/deduplication";
+ url = adminUrl + "admin/v2/namespaces/public/dedup-2/deduplication";
res = makePostRequest(url, "true");
ASSERT_TRUE(res == 204 || res == 409);
diff --git a/pulsar-client-cpp/tests/ClientTest.cc
b/pulsar-client-cpp/tests/ClientTest.cc
index 5bc0246a3f..d4eaca794c 100644
--- a/pulsar-client-cpp/tests/ClientTest.cc
+++ b/pulsar-client-cpp/tests/ClientTest.cc
@@ -23,7 +23,7 @@
using namespace pulsar;
-static std::string lookupUrl = "pulsar://localhost:8885";
+static std::string lookupUrl = "pulsar://localhost:6650";
TEST(ClientTest, testChecksumComputation) {
std::string data = "test";
diff --git a/pulsar-client-cpp/tests/ConsumerConfigurationTest.cc
b/pulsar-client-cpp/tests/ConsumerConfigurationTest.cc
index cede638ec0..e4944ec6f5 100644
--- a/pulsar-client-cpp/tests/ConsumerConfigurationTest.cc
+++ b/pulsar-client-cpp/tests/ConsumerConfigurationTest.cc
@@ -25,8 +25,8 @@
using namespace pulsar;
TEST(ConsumerConfigurationTest, testReadCompactPersistentExclusive) {
- std::string lookupUrl = "pulsar://localhost:8885";
- std::string topicName = "persistent://prop/unit/ns1/persist-topic";
+ std::string lookupUrl = "pulsar://localhost:6650";
+ std::string topicName = "persist-topic";
std::string subName = "test-persist-exclusive";
Result result;
@@ -45,8 +45,8 @@ TEST(ConsumerConfigurationTest,
testReadCompactPersistentExclusive) {
}
TEST(ConsumerConfigurationTest, testReadCompactPersistentFailover) {
- std::string lookupUrl = "pulsar://localhost:8885";
- std::string topicName = "persistent://prop/unit/ns1/persist-topic";
+ std::string lookupUrl = "pulsar://localhost:6650";
+ std::string topicName = "persist-topic";
std::string subName = "test-persist-fail-over";
Result result;
@@ -65,8 +65,8 @@ TEST(ConsumerConfigurationTest,
testReadCompactPersistentFailover) {
}
TEST(ConsumerConfigurationTest, testReadCompactPersistentShared) {
- std::string lookupUrl = "pulsar://localhost:8885";
- std::string topicName = "persistent://prop/unit/ns1/persist-topic";
+ std::string lookupUrl = "pulsar://localhost:6650";
+ std::string topicName = "persist-topic";
std::string subName = "test-persist-shared";
Result result;
@@ -85,8 +85,8 @@ TEST(ConsumerConfigurationTest,
testReadCompactPersistentShared) {
}
TEST(ConsumerConfigurationTest, testReadCompactNonPersistentExclusive) {
- std::string lookupUrl = "pulsar://localhost:8885";
- std::string topicName =
"non-persistent://prop/unit/ns1/testNonPersistentTopic";
+ std::string lookupUrl = "pulsar://localhost:6650";
+ std::string topicName =
"non-persistent://public/default/testNonPersistentTopic";
std::string subName = "test-non-persist-exclusive";
Result result;
diff --git a/pulsar-client-cpp/tests/ConsumerStatsTest.cc
b/pulsar-client-cpp/tests/ConsumerStatsTest.cc
index 684302733a..585d664fbd 100644
--- a/pulsar-client-cpp/tests/ConsumerStatsTest.cc
+++ b/pulsar-client-cpp/tests/ConsumerStatsTest.cc
@@ -36,8 +36,8 @@ DECLARE_LOG_OBJECT();
using namespace pulsar;
-static std::string lookupUrl = "http://localhost:8765";;
-static std::string adminUrl = "http://localhost:8765/";;
+static std::string lookupUrl = "pulsar://localhost:6650";
+static std::string adminUrl = "http://localhost:8080/";;
void partitionedCallbackFunction(Result result, BrokerConsumerStats
brokerConsumerStats, long expectedBacklog,
Latch& latch, int index) {
@@ -60,7 +60,7 @@ TEST(ConsumerStatsTest, testBacklogInfo) {
long epochTime = time(NULL);
std::string testName = "testBacklogInfo-" +
boost::lexical_cast<std::string>(epochTime);
Client client(lookupUrl);
- std::string topicName = "persistent://property/cluster/namespace/" +
testName;
+ std::string topicName = "persistent://public/default/" + testName;
std::string subName = "subscription-name";
ConsumerConfiguration conf;
conf.setBrokerConsumerStatsCacheTimeInMs(3 * 1000);
@@ -115,7 +115,7 @@ TEST(ConsumerStatsTest, testFailure) {
long epochTime = time(NULL);
std::string testName = "testFailure-" +
boost::lexical_cast<std::string>(epochTime);
Client client(lookupUrl);
- std::string topicName = "persistent://property/cluster/namespace/" +
testName;
+ std::string topicName = "persistent://public/default/" + testName;
std::string subName = "subscription-name";
Consumer consumer;
Promise<Result, Consumer> consumerPromise;
@@ -160,7 +160,7 @@ TEST(ConsumerStatsTest, testCachingMechanism) {
long epochTime = time(NULL);
std::string testName = "testCachingMechanism-" +
boost::lexical_cast<std::string>(epochTime);
Client client(lookupUrl);
- std::string topicName = "persistent://property/cluster/namespace/" +
testName;
+ std::string topicName = "persistent://public/default/" + testName;
std::string subName = "subscription-name";
ConsumerConfiguration conf;
conf.setBrokerConsumerStatsCacheTimeInMs(3.5 * 1000);
@@ -235,11 +235,11 @@ TEST(ConsumerStatsTest, testAsyncCallOnPartitionedTopic) {
long epochTime = time(NULL);
std::string testName = "testAsyncCallOnPartitionedTopic-" +
boost::lexical_cast<std::string>(epochTime);
Client client(lookupUrl);
- std::string topicName = "persistent://property/cluster/namespace/" +
testName;
+ std::string topicName = "persistent://public/default/" + testName;
std::string subName = "subscription-name";
// call admin api to create partitioned topics
- std::string url = adminUrl +
"admin/persistent/property/cluster/namespace/" + testName + "/partitions";
+ std::string url = adminUrl + "admin/v2/persistent/public/default/" +
testName + "/partitions";
int res = makePutRequest(url, "7");
LOG_INFO("res = " << res);
diff --git a/pulsar-client-cpp/tests/ReaderTest.cc
b/pulsar-client-cpp/tests/ReaderTest.cc
index fd6cf5d8a5..0ca1f751d3 100644
--- a/pulsar-client-cpp/tests/ReaderTest.cc
+++ b/pulsar-client-cpp/tests/ReaderTest.cc
@@ -28,12 +28,12 @@ DECLARE_LOG_OBJECT()
using namespace pulsar;
-static std::string serviceUrl = "pulsar://localhost:8885";
+static std::string serviceUrl = "pulsar://localhost:6650";
TEST(ReaderTest, testSimpleReader) {
Client client(serviceUrl);
- std::string topicName =
"persistent://property/cluster/namespace/test-simple-reader";
+ std::string topicName = "persistent://public/default/test-simple-reader";
ReaderConfiguration readerConf;
Reader reader;
@@ -65,7 +65,7 @@ TEST(ReaderTest, testSimpleReader) {
TEST(ReaderTest, testReaderAfterMessagesWerePublished) {
Client client(serviceUrl);
- std::string topicName =
"persistent://property/cluster/namespace/testReaderAfterMessagesWerePublished";
+ std::string topicName =
"persistent://public/default/testReaderAfterMessagesWerePublished";
Producer producer;
ASSERT_EQ(ResultOk, client.createProducer(topicName, producer));
@@ -97,7 +97,7 @@ TEST(ReaderTest, testReaderAfterMessagesWerePublished) {
TEST(ReaderTest, testMultipleReaders) {
Client client(serviceUrl);
- std::string topicName =
"persistent://property/cluster/namespace/testMultipleReaders";
+ std::string topicName = "persistent://public/default/testMultipleReaders";
Producer producer;
ASSERT_EQ(ResultOk, client.createProducer(topicName, producer));
@@ -142,7 +142,7 @@ TEST(ReaderTest, testMultipleReaders) {
TEST(ReaderTest, testReaderOnLastMessage) {
Client client(serviceUrl);
- std::string topicName =
"persistent://property/cluster/namespace/testReaderOnLastMessage";
+ std::string topicName =
"persistent://public/default/testReaderOnLastMessage";
Producer producer;
ASSERT_EQ(ResultOk, client.createProducer(topicName, producer));
@@ -180,7 +180,7 @@ TEST(ReaderTest, testReaderOnLastMessage) {
TEST(ReaderTest, testReaderOnSpecificMessage) {
Client client(serviceUrl);
- std::string topicName =
"persistent://property/cluster/namespace/testReaderOnSpecificMessage";
+ std::string topicName =
"persistent://public/default/testReaderOnSpecificMessage";
Producer producer;
ASSERT_EQ(ResultOk, client.createProducer(topicName, producer));
@@ -231,7 +231,7 @@ TEST(ReaderTest, testReaderOnSpecificMessage) {
TEST(ReaderTest, testReaderOnSpecificMessageWithBatches) {
Client client(serviceUrl);
- std::string topicName =
"persistent://property/cluster/namespace/testReaderOnSpecificMessageWithBatches";
+ std::string topicName =
"persistent://public/default/testReaderOnSpecificMessageWithBatches";
Producer producer;
// Enable batching
@@ -291,7 +291,7 @@ TEST(ReaderTest, testReaderOnSpecificMessageWithBatches) {
TEST(ReaderTest, testReaderReachEndOfTopic) {
Client client(serviceUrl);
- std::string topicName =
"persistent://property/cluster/namespace/testReaderReachEndOfTopic";
+ std::string topicName =
"persistent://public/default/testReaderReachEndOfTopic";
// 1. create producer
Producer producer;
@@ -366,8 +366,7 @@ TEST(ReaderTest, testReaderReachEndOfTopic) {
TEST(ReaderTest, testReaderReachEndOfTopicMessageWithoutBatches) {
Client client(serviceUrl);
- std::string topicName =
-
"persistent://property/cluster/namespace/testReaderReachEndOfTopicMessageWithBatches";
+ std::string topicName =
"persistent://public/default/testReaderReachEndOfTopicMessageWithBatches";
// 1. create producer
Producer producer;
diff --git a/pulsar-client-cpp/tests/ZeroQueueSizeTest.cc
b/pulsar-client-cpp/tests/ZeroQueueSizeTest.cc
index e1cd1fcadf..890cee602e 100644
--- a/pulsar-client-cpp/tests/ZeroQueueSizeTest.cc
+++ b/pulsar-client-cpp/tests/ZeroQueueSizeTest.cc
@@ -26,7 +26,7 @@ using namespace pulsar;
static int totalMessages = 10;
static int globalCount = 0;
-static std::string lookupUrl = "pulsar://localhost:8885";
+static std::string lookupUrl = "pulsar://localhost:6650";
static std::string contentBase = "msg-";
static void messageListenerFunction(Consumer consumer, const Message& msg,
Latch& latch) {
@@ -41,7 +41,7 @@ static void messageListenerFunction(Consumer consumer, const
Message& msg, Latch
TEST(ZeroQueueSizeTest, testProduceConsume) {
Client client(lookupUrl);
- std::string topicName = "persistent://prop/unit/ns1/zero-queue-size";
+ std::string topicName = "zero-queue-size";
std::string subName = "my-sub-name";
Producer producer;
@@ -80,7 +80,7 @@ TEST(ZeroQueueSizeTest, testProduceConsume) {
TEST(ZeroQueueSizeTest, testMessageListener) {
Client client(lookupUrl);
- std::string topicName =
"persistent://prop/unit/ns/zero-queue-size-listener";
+ std::string topicName = "zero-queue-size-listener";
std::string subName = "my-sub-name";
Producer producer;
diff --git
a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java
b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java
index ab9686c363..8a1e258290 100644
---
a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java
+++
b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java
@@ -113,10 +113,6 @@
"--auth_params" }, description = "Authentication parameters,
e.g., \"key1:val1,key2:val2\"")
public String authParams;
- @Parameter(names = {
- "--use-tls" }, description = "Use TLS encryption on the
connection")
- public boolean useTls;
-
@Parameter(names = {
"--trust-cert-file" }, description = "Path for the trusted TLS
certificate file")
public String tlsTrustCertsFilePath = "";
@@ -178,10 +174,6 @@ public static void main(String[] args) throws Exception {
arguments.authParams = prop.getProperty("authParams", null);
}
- if (arguments.useTls == false) {
- arguments.useTls =
Boolean.parseBoolean(prop.getProperty("useTls"));
- }
-
if (isBlank(arguments.tlsTrustCertsFilePath)) {
arguments.tlsTrustCertsFilePath =
prop.getProperty("tlsTrustCertsFilePath", "");
}
@@ -218,7 +210,6 @@ public static void main(String[] args) throws Exception {
.connectionsPerBroker(arguments.maxConnections) //
.statsInterval(arguments.statsIntervalSeconds,
TimeUnit.SECONDS) //
.ioThreads(Runtime.getRuntime().availableProcessors()) //
- .enableTls(arguments.useTls) //
.tlsTrustCertsFilePath(arguments.tlsTrustCertsFilePath);
if (isNotBlank(arguments.authPluginClassName)) {
clientBuilder.authentication(arguments.authPluginClassName,
arguments.authParams);
diff --git
a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java
b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java
index bfd4fbbbbf..6967d43143 100644
---
a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java
+++
b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java
@@ -139,10 +139,6 @@
@Parameter(names = "--warmup-time", description = "Warm-up time in
seconds (Default: 1 sec)")
public double warmupTimeSeconds = 1.0;
- @Parameter(names = {
- "--use-tls" }, description = "Use TLS encryption on the
connection")
- public boolean useTls;
-
@Parameter(names = {
"--trust-cert-file" }, description = "Path for the trusted TLS
certificate file")
public String tlsTrustCertsFilePath = "";
@@ -206,10 +202,6 @@ public static void main(String[] args) throws Exception {
arguments.authParams = prop.getProperty("authParams", null);
}
- if (arguments.useTls == false) {
- arguments.useTls =
Boolean.parseBoolean(prop.getProperty("useTls"));
- }
-
if (isBlank(arguments.tlsTrustCertsFilePath)) {
arguments.tlsTrustCertsFilePath =
prop.getProperty("tlsTrustCertsFilePath", "");
}
@@ -237,7 +229,6 @@ public static void main(String[] args) throws Exception {
.connectionsPerBroker(arguments.maxConnections) //
.ioThreads(Runtime.getRuntime().availableProcessors()) //
.statsInterval(arguments.statsIntervalSeconds,
TimeUnit.SECONDS) //
- .enableTls(arguments.useTls) //
.tlsTrustCertsFilePath(arguments.tlsTrustCertsFilePath);
if (isNotBlank(arguments.authPluginClassName)) {
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on 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
